diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 937b14c..1840665 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -53,6 +53,29 @@ jobs: build/Testing/Temporary/LastTest.log if-no-files-found: ignore + linux-mdbx: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + submodules: true + - run: git submodule update --init --recursive + - name: Configure MDBX + run: cmake -S . -B build-mdbx -DLOGIT_CPP_BUILD_TESTS=ON -DLOGIT_CPP_BUILD_EXAMPLES=ON -DLOGIT_WITH_MDBX=ON -DLOGIT_USE_SUBMODULES=ON -DCMAKE_CXX_STANDARD=17 + - name: Build MDBX + run: cmake --build build-mdbx + - name: Test MDBX + run: ctest --test-dir build-mdbx --output-on-failure -R mdbx_logger_test + - name: Upload logs + if: failure() + uses: actions/upload-artifact@v4 + with: + name: logs-mdbx + path: | + build-mdbx/CMakeFiles/CMakeOutput.log + build-mdbx/Testing/Temporary/LastTest.log + if-no-files-found: ignore + windows: runs-on: windows-latest strategy: diff --git a/.gitmodules b/.gitmodules index 9f4474f..659e30a 100644 --- a/.gitmodules +++ b/.gitmodules @@ -13,3 +13,9 @@ [submodule "external/zlib"] path = external/zlib url = https://github.com/madler/zlib +[submodule "external/mdbx-containers"] + path = external/mdbx-containers + url = https://github.com/NewYaroslav/mdbx-containers.git +[submodule "external/kurlyk"] + path = external/kurlyk + url = https://github.com/NewYaroslav/kurlyk.git diff --git a/CMakeLists.txt b/CMakeLists.txt index 66a7c8d..2efdaba 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -12,6 +12,7 @@ option(LOGIT_WITH_CONTEXT "Enable MDC/NDC diagnostic context support" OFF) option(LOGIT_WITH_OTLP "Enable OTLP/HTTP log export via optional kurlyk dependency" OFF) option(LOGIT_WITH_PROMETHEUS "Enable Prometheus text payload support" OFF) option(LOGIT_WITH_PROMETHEUS_SERVER "Enable Prometheus HTTP server backend" OFF) +option(LOGIT_WITH_MDBX "Enable MDBX structured log storage backend" OFF) option(LOGIT_USE_SUBMODULES "Allow bundled optional dependency fallback" OFF) option(LOGIT_WITH_SYSLOG "Enable POSIX syslog backend" ON) option(LOGIT_WITH_WIN_EVENT_LOG "Enable Windows Event Log backend" ON) @@ -24,7 +25,7 @@ option(LOGIT_USE_MPSC_RING "Enable lock-free TaskExecutor queue" ON) option(LOGIT_ENABLE_DROP_OLDEST_SLOWPATH "Enable TaskExecutor DropOldest slow-path" ON) if(NOT DEFINED CMAKE_CXX_STANDARD) - if(LOGIT_WITH_OTLP OR LOGIT_WITH_PROMETHEUS_SERVER) + if(LOGIT_WITH_OTLP OR LOGIT_WITH_PROMETHEUS_SERVER OR LOGIT_WITH_MDBX) set(CMAKE_CXX_STANDARD 17) else() set(CMAKE_CXX_STANDARD 11) @@ -32,8 +33,12 @@ if(NOT DEFINED CMAKE_CXX_STANDARD) endif() set(CMAKE_CXX_STANDARD_REQUIRED ON) +if(LOGIT_WITH_MDBX AND CMAKE_CXX_STANDARD LESS 17) + message(FATAL_ERROR "LOGIT_WITH_MDBX requires C++17 or newer.") +endif() + # Dependency: TimeShield -find_package(TimeShield 1.0.4 QUIET CONFIG) +find_package(TimeShield 1.0.6 QUIET CONFIG) if(NOT TimeShield_FOUND) if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/external/time-shield-cpp/CMakeLists.txt") add_subdirectory(external/time-shield-cpp) @@ -185,6 +190,40 @@ if(LOGIT_WITH_PROMETHEUS_SERVER) endif() endif() +# ---------- MDBX ---------- +if(LOGIT_WITH_MDBX) + if(EMSCRIPTEN) + message(FATAL_ERROR "LOGIT_WITH_MDBX is not supported for Emscripten.") + endif() + if(MSVC) + message(FATAL_ERROR "LOGIT_WITH_MDBX is not supported with MSVC until mdbx-containers supports MSVC.") + endif() + + if(NOT TARGET mdbx_containers::mdbx_containers) + find_package(mdbx_containers QUIET CONFIG) + endif() + + if(NOT TARGET mdbx_containers::mdbx_containers AND LOGIT_USE_SUBMODULES) + if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/external/mdbx-containers/CMakeLists.txt") + set(MDBXC_BUILD_TESTS OFF CACHE BOOL "" FORCE) + set(MDBXC_BUILD_EXAMPLES OFF CACHE BOOL "" FORCE) + set(MDBXC_USE_ASAN OFF CACHE BOOL "" FORCE) + set(MDBXC_DEPS_MODE BUNDLED CACHE STRING "Dependency mode for MDBX: AUTO|SYSTEM|BUNDLED" FORCE) + add_subdirectory(external/mdbx-containers EXCLUDE_FROM_ALL) + endif() + endif() + + if(NOT TARGET mdbx_containers::mdbx_containers) + message(FATAL_ERROR "mdbx-containers not found. Install it or add it as external/mdbx-containers and enable LOGIT_USE_SUBMODULES.") + endif() + + target_compile_definitions(log-it-cpp INTERFACE LOGIT_WITH_MDBX=1) + target_link_libraries(log-it-cpp INTERFACE + $ + $ + ) +endif() + # ---------- GZIP (zlib) ---------- if(LOGIT_WITH_GZIP) if(NOT TARGET ZLIB::ZLIB) @@ -248,27 +287,38 @@ install(DIRECTORY include/ DESTINATION include) install(TARGETS log-it-cpp EXPORT log-it-cppTargets) +set(_logit_install_export_supported ON) + # install(EXPORT) requires all linked targets to be in an export set. -# When kurlyk is pre-installed (IMPORTED target) the export works. -# When kurlyk is a submodule (non-IMPORTED), packaging is unsupported. -# The FATAL_ERROR is deferred to install time so development builds with -# submodules still work; only `cmake --install` is blocked. +# When optional dependencies are pre-installed (IMPORTED targets), the +# export works. When they are submodules (non-IMPORTED), packaging is +# unsupported. The FATAL_ERROR is deferred to install time so development +# builds with submodules still work; only `cmake --install` is blocked. if(LOGIT_WITH_OTLP AND TARGET kurlyk) get_target_property(_kurlyk_imported kurlyk IMPORTED) if(NOT _kurlyk_imported) + set(_logit_install_export_supported OFF) install(CODE [[ message(FATAL_ERROR "log-it-cpp: Installing with LOGIT_WITH_OTLP=ON and bundled kurlyk is not supported. " "Install kurlyk separately and use find_package(kurlyk), or disable LOGIT_WITH_OTLP for install.") ]]) - else() - install(EXPORT log-it-cppTargets - FILE log-it-cppTargets.cmake - NAMESPACE log-it-cpp:: - DESTINATION lib/cmake/log-it-cpp - ) endif() -else() +endif() + +if(LOGIT_WITH_MDBX AND TARGET mdbx_containers::mdbx_containers) + get_target_property(_mdbx_containers_imported mdbx_containers::mdbx_containers IMPORTED) + if(NOT _mdbx_containers_imported) + set(_logit_install_export_supported OFF) + install(CODE [[ + message(FATAL_ERROR + "log-it-cpp: Installing with LOGIT_WITH_MDBX=ON and bundled mdbx-containers is not supported. " + "Install mdbx-containers separately and use find_package(mdbx_containers), or disable LOGIT_WITH_MDBX for install.") + ]]) + endif() +endif() + +if(_logit_install_export_supported) install(EXPORT log-it-cppTargets FILE log-it-cppTargets.cmake NAMESPACE log-it-cpp:: diff --git a/cmake/log-it-cppConfig.cmake.in b/cmake/log-it-cppConfig.cmake.in index b4190ca..f0a8663 100644 --- a/cmake/log-it-cppConfig.cmake.in +++ b/cmake/log-it-cppConfig.cmake.in @@ -2,5 +2,8 @@ include(CMakeFindDependencyMacro) find_dependency(TimeShield) +if(@LOGIT_WITH_MDBX@) + find_dependency(mdbx_containers) +endif() include("${CMAKE_CURRENT_LIST_DIR}/log-it-cppTargets.cmake") diff --git a/examples/example_logit_mdbx_logger.cpp b/examples/example_logit_mdbx_logger.cpp new file mode 100644 index 0000000..d8c36d2 --- /dev/null +++ b/examples/example_logit_mdbx_logger.cpp @@ -0,0 +1,252 @@ +/// \file example_logit_mdbx_logger.cpp +/// \brief Demonstrates structured log storage with MdbxLogger, including macros, +/// querying by date range, and reading sessions/payloads. + +// #define LOGIT_BASE_PATH "E:\\_repoz\\log-it-cpp" <- set via CMake + +#include + +#ifdef LOGIT_WITH_MDBX + +#include +#include +#include +#include +#include +#include +#include + +namespace { + +std::string make_db_path() { + std::ostringstream os; + os << logit::get_exec_dir() << "/example_mdbx_logger_" << LOGIT_CURRENT_TIMESTAMP_MS() << ".mdbx"; + return os.str(); +} + +void cleanup_db(const std::string& path) { + std::remove(path.c_str()); + std::remove((path + "-lck").c_str()); +} + +} // namespace + +int main() { + const std::string db_path = make_db_path(); + cleanup_db(db_path); + + std::cout << "MdbxLogger example starting, db=" << db_path << std::endl; + + // ------------------------------------------------------------------ + // 1. Configure and add the MDBX backend + // ------------------------------------------------------------------ + logit::MdbxLogger::Config mdbx_config; + mdbx_config.path = db_path; + mdbx_config.app_name = "example-mdbx"; + mdbx_config.async = true; + mdbx_config.max_queue_size = 256; + mdbx_config.max_batch_size = 32; + mdbx_config.large_payload_threshold = 256; + mdbx_config.store_large_payloads_separately = true; + + // Optional: capture write errors instead of writing to stderr + mdbx_config.on_error = [](const std::string& msg) { + std::cerr << "[MdbxLogger error callback] " << msg << std::endl; + }; + + // Add to the registry in single_mode so it does not duplicate console output. + // After this call the logger index depends on what was added before it. + LOGIT_ADD_LOGGER_SINGLE_MODE( + logit::MdbxLogger, + (mdbx_config), + logit::SimpleLogFormatter, + ("[%l] %v")); + + // The MDBX logger is now the last added backend; its index is: + const int mdbx_index = static_cast(logit::Logger::get_instance().logger_count()) - 1; + + // Also add a console logger for live observation (optional). + LOGIT_ADD_CONSOLE_DEFAULT(); + + // ------------------------------------------------------------------ + // 2. Log messages via standard macros + // ------------------------------------------------------------------ + LOGIT_INFO("Application started"); + LOGIT_WARN("Disk usage is above 80%%"); + + { + LOGIT_SCOPE_INFO("process_batch"); + LOGIT_INFO("Processing 42 items"); + LOGIT_DEBUG("Detail: item id=7, status=pending"); + + // This message is larger than large_payload_threshold and will spill + // into the separate log_payloads table with a preview kept inline. + std::string big; + big.reserve(512); + for (int i = 0; i < 50; ++i) { + big += "chunk-" + std::to_string(i) + " "; + } + LOGIT_INFO(big); + } + + LOGIT_ERROR("Connection timeout to upstream-3"); + LOGIT_FATAL("Critical: unable to recover transaction state"); + + // Wait until the async queue is flushed. + LOGIT_WAIT(); + + // ------------------------------------------------------------------ + // 3. Read the MDBX logger directly via typed macro helpers + // ------------------------------------------------------------------ + LOGIT_WITH_LOGGER_AS(mdbx_index, logit::MdbxLogger, mdbx) { + // Session metadata + auto session_opt = mdbx->read_session(mdbx->session_id()); + if (session_opt) { + std::cout << "\n--- Session ---" << std::endl; + std::cout << " app_name: " << session_opt->app_name << std::endl; + std::cout << " process_id: " << session_opt->process_id << std::endl; + std::cout << " started_ms: " << session_opt->start_time_ms << std::endl; + std::cout << " schema_ver: " << session_opt->schema_version << std::endl; + } + + // All records in a wide time window (last 24 hours). + const int64_t now_ms = LOGIT_CURRENT_TIMESTAMP_MS(); + auto all_records = mdbx->read_range(now_ms - 24 * 60 * 60 * 1000, now_ms + 1); + + std::cout << "\n--- All records (" << all_records.size() << ") ---" << std::endl; + for (const auto& r : all_records) { + std::cout << " [" << logit::to_string(r.level) << "] " + << r.timestamp_ms << " seq=" << r.sequence + << " msg=\"" << r.message << "\""; + if (r.payload_id != 0) { + std::cout << " [payload_id=" << r.payload_id << "]"; + } + std::cout << std::endl; + } + + // Level-based client-side filter + std::cout << "\n--- Records with level >= WARN ---" << std::endl; + for (const auto& r : all_records) { + if (static_cast(r.level) >= static_cast(logit::LogLevel::LOG_LVL_WARN)) { + std::cout << " [" << logit::to_string(r.level) << "] " + << r.timestamp_ms << " " << r.message << std::endl; + } + } + + // Spilled payloads (decompressed transparently) + std::cout << "\n--- Spilled payloads ---" << std::endl; + for (const auto& r : all_records) { + if (r.payload_id != 0) { + auto data_opt = mdbx->read_payload_data(r.payload_id); + if (data_opt) { + std::cout << " payload_id=" << r.payload_id + << " size=" << data_opt->size() + << " preview=\"" << r.message << "\"" + << std::endl; + } else { + std::cout << " payload_id=" << r.payload_id << " FAILED to read/decompress" + << std::endl; + } + } + } + + // Query by today's local midnight + try { + auto now_local = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now()); + std::tm* tm_now = std::localtime(&now_local); + std::tm tm_midnight = *tm_now; + tm_midnight.tm_hour = 0; + tm_midnight.tm_min = 0; + tm_midnight.tm_sec = 0; + std::time_t midnight_t = std::mktime(&tm_midnight); + + auto today_ms = std::chrono::milliseconds(static_cast(midnight_t) * 1000); + auto tomorrow_ms = today_ms + std::chrono::hours(24); + + auto day_records = mdbx->read_range( + today_ms.count(), + tomorrow_ms.count()); + + std::cout << "\n--- Records for today (" << day_records.size() << ") ---" << std::endl; + for (const auto& r : day_records) { + std::cout << " [" << logit::to_string(r.level) << "] " + << r.timestamp_ms << " " << r.message << std::endl; + } + } catch (const std::exception& e) { + std::cerr << "Date query example skipped: " << e.what() << std::endl; + } + + // read_recent: last 100 records in ascending order + auto recent = mdbx->read_recent(100, 0, logit::LogReadOrder::Ascending); + std::cout << "\n--- read_recent(100) ascending (" << recent.size() << ") ---" << std::endl; + for (const auto& r : recent) { + std::cout << " [" << logit::to_string(r.level) << "] " + << r.timestamp_ms << " " << r.message << std::endl; + } + + // Live subscription: snapshot + real-time updates + std::vector live_updates; + uint64_t cb_id = mdbx->add_log_callback( + [&live_updates](const logit::LogRecordView& v) { + live_updates.push_back(v); + }); + + LOGIT_INFO("Live event 1 via callback"); + LOGIT_INFO("Live event 2 via callback"); + LOGIT_WAIT(); + + std::cout << "\n--- Live updates received (" << live_updates.size() + << ") ---" << std::endl; + for (const auto& r : live_updates) { + std::cout << " [" << logit::to_string(r.level) << "] " + << r.message << std::endl; + } + + if (mdbx->remove_log_callback(cb_id)) { + std::cout << "Callback removed" << std::endl; + } + + LOGIT_INFO("Event after unsubscribe"); + LOGIT_WAIT(); + std::cout << "Live updates after unsubscribe: " << live_updates.size() + << std::endl; + + // Statistics + std::cout << "\n--- Statistics ---" << std::endl; + std::cout << " dropped: " << mdbx->dropped_count() << std::endl; + std::cout << " failed_writes: " << mdbx->failed_export_count() << std::endl; + } else { + std::cerr << "MDBX logger not found at index " << mdbx_index << std::endl; + LOGIT_SHUTDOWN(); + cleanup_db(db_path); + return 1; + } + + // ------------------------------------------------------------------ + // 5. Graceful shutdown and cleanup + // ------------------------------------------------------------------ + LOGIT_SHUTDOWN(); + + // After shutdown the session end_time_ms is persisted. + LOGIT_WITH_LOGGER_AS(mdbx_index, logit::MdbxLogger, mdbx) { + auto session_opt = mdbx->read_session(mdbx->session_id()); + if (session_opt) { + std::cout << " session end_ms: " << session_opt->end_time_ms << std::endl; + } + } + + cleanup_db(db_path); + std::cout << "\nMdbxLogger example completed." << std::endl; + return 0; +} + +#else + +#include + +int main() { + std::cout << "MdbxLogger example requires LOGIT_WITH_MDBX=ON." << std::endl; + return 0; +} + +#endif diff --git a/external/kurlyk b/external/kurlyk new file mode 160000 index 0000000..f683f0b --- /dev/null +++ b/external/kurlyk @@ -0,0 +1 @@ +Subproject commit f683f0bf2249cef5236f2f591adff1061afe7080 diff --git a/external/mdbx-containers b/external/mdbx-containers new file mode 160000 index 0000000..fac4001 --- /dev/null +++ b/external/mdbx-containers @@ -0,0 +1 @@ +Subproject commit fac400198b4e121db63df47fd908c4e30d5723ef diff --git a/external/time-shield-cpp b/external/time-shield-cpp index 23f8715..d3c251b 160000 --- a/external/time-shield-cpp +++ b/external/time-shield-cpp @@ -1 +1 @@ -Subproject commit 23f8715459e3e7f537657f0d029b5427b4062a36 +Subproject commit d3c251bf173ee6222f6e79186a34c4a0592d5767 diff --git a/include/logit_cpp/logit/Logger.hpp b/include/logit_cpp/logit/Logger.hpp index dcd3a66..5b48916 100644 --- a/include/logit_cpp/logit/Logger.hpp +++ b/include/logit_cpp/logit/Logger.hpp @@ -13,6 +13,7 @@ #include #include #include +#include #if __cplusplus >= 201703L #include @@ -398,6 +399,21 @@ namespace logit { } } + /// \brief Returns the number of registered logger strategies. + std::size_t logger_count() const { + LoggerReadLock lock(m_loggers_mx); + return m_loggers.size(); + } + + /// \brief Retrieves a typed backend pointer from a logger by index. + template + LoggerT* get_logger_as(int logger_index) const { + auto strategy = get_strategy_snapshot(logger_index); + return (strategy && strategy->logger) + ? dynamic_cast(strategy->logger.get()) + : nullptr; + } + /// \brief Shuts down logger system. /// /// Disables further logging, waits for asynchronous tasks to complete, diff --git a/include/logit_cpp/logit/detail/CompressionUtils.hpp b/include/logit_cpp/logit/detail/CompressionUtils.hpp new file mode 100644 index 0000000..d448848 --- /dev/null +++ b/include/logit_cpp/logit/detail/CompressionUtils.hpp @@ -0,0 +1,189 @@ +#pragma once +#ifndef _LOGIT_DETAIL_COMPRESSION_UTILS_HPP_INCLUDED +#define _LOGIT_DETAIL_COMPRESSION_UTILS_HPP_INCLUDED + +/// \file CompressionUtils.hpp +/// \brief Shared gzip/zstd compression helpers used by OTLP and MDBX backends. + +#include +#include + +#if defined(LOGIT_HAS_ZLIB) +# include +#endif + +#if defined(LOGIT_HAS_ZSTD) +# include +#endif + +namespace logit { +namespace detail { + +/// \brief Compress a string with gzip. +/// \param input Uncompressed data. +/// \param[out] output Compressed result (valid only on success). +/// \param level Compression level 1-9. +/// \return true on success, false if zlib is unavailable or compression fails. +/// \note Callers must check the return value and handle fallback explicitly. +inline bool compress_string_gzip(const std::string& input, std::string& output, int level) { +#if defined(LOGIT_HAS_ZLIB) + z_stream zs; + zs.zalloc = Z_NULL; + zs.zfree = Z_NULL; + zs.opaque = Z_NULL; + zs.next_in = reinterpret_cast(const_cast(input.data())); + zs.avail_in = static_cast(input.size()); + zs.next_out = Z_NULL; + zs.avail_out = 0; + + if (level < 1) level = 1; + if (level > 9) level = 9; + + int window_bits = 15 + 16; + if (deflateInit2(&zs, level, Z_DEFLATED, window_bits, 8, Z_DEFAULT_STRATEGY) != Z_OK) { + return false; + } + + output.clear(); + + int ret = Z_OK; + std::size_t offset = 0; + const std::size_t buf_size = 32768; + + do { + output.resize(offset + buf_size); + zs.next_out = reinterpret_cast(&output[offset]); + zs.avail_out = static_cast(buf_size); + + ret = deflate(&zs, Z_FINISH); + if (ret == Z_STREAM_ERROR) { + deflateEnd(&zs); + return false; + } + + offset = zs.total_out; + } while (ret != Z_STREAM_END); + + output.resize(zs.total_out); + deflateEnd(&zs); + return true; +#else + (void)input; (void)output; (void)level; + return false; +#endif +} + +/// \brief Decompress gzip-compressed bytes. +/// \param input Compressed data. +/// \param[out] output Decompressed result (valid only on success). +/// \return true on success, false if zlib is unavailable or decompression fails. +inline bool decompress_string_gzip(const std::string& input, std::string& output) { +#if defined(LOGIT_HAS_ZLIB) + z_stream zs; + zs.zalloc = Z_NULL; + zs.zfree = Z_NULL; + zs.opaque = Z_NULL; + zs.next_in = reinterpret_cast(const_cast(input.data())); + zs.avail_in = static_cast(input.size()); + zs.next_out = Z_NULL; + zs.avail_out = 0; + + int window_bits = 15 + 32; // automatic header detection (gzip/zlib) + if (inflateInit2(&zs, window_bits) != Z_OK) { + return false; + } + + output.clear(); + + int ret = Z_OK; + std::size_t offset = 0; + const std::size_t buf_size = 32768; + + do { + output.resize(offset + buf_size); + zs.next_out = reinterpret_cast(&output[offset]); + zs.avail_out = static_cast(buf_size); + + ret = inflate(&zs, Z_NO_FLUSH); + if (ret == Z_STREAM_ERROR || ret == Z_DATA_ERROR || ret == Z_MEM_ERROR) { + inflateEnd(&zs); + return false; + } + + offset = zs.total_out; + } while (ret != Z_STREAM_END); + + output.resize(zs.total_out); + inflateEnd(&zs); + return true; +#else + (void)input; (void)output; + return false; +#endif +} + +/// \brief Compress a string with zstd. +/// \param input Uncompressed data. +/// \param[out] output Compressed result (valid only on success). +/// \param level Compression level 1-19. +/// \return true on success, false if zstd is unavailable or compression fails. +/// \note Callers must check the return value and handle fallback explicitly. +inline bool compress_string_zstd(const std::string& input, std::string& output, int level) { +#if defined(LOGIT_HAS_ZSTD) + if (level < 1) level = 1; + if (level > 19) level = 19; + + std::size_t bound = ZSTD_compressBound(input.size()); + output.resize(bound); + + std::size_t result = ZSTD_compress( + &output[0], output.size(), + input.data(), input.size(), + level); + + if (ZSTD_isError(result)) { + output.clear(); + return false; + } + + output.resize(result); + return true; +#else + (void)input; (void)output; (void)level; + return false; +#endif +} + +/// \brief Decompress zstd-compressed bytes. +/// \param input Compressed data. +/// \param[out] output Decompressed result (valid only on success). +/// \return true on success, false if zstd is unavailable or decompression fails. +inline bool decompress_string_zstd(const std::string& input, std::string& output) { +#if defined(LOGIT_HAS_ZSTD) + std::size_t const d_size = ZSTD_getFrameContentSize(input.data(), input.size()); + if (d_size == ZSTD_CONTENTSIZE_ERROR || d_size == ZSTD_CONTENTSIZE_UNKNOWN) { + return false; + } + + output.resize(d_size); + std::size_t const result = ZSTD_decompress( + &output[0], output.size(), + input.data(), input.size()); + + if (ZSTD_isError(result)) { + output.clear(); + return false; + } + + output.resize(result); + return true; +#else + (void)input; (void)output; + return false; +#endif +} + +} // namespace detail +} // namespace logit + +#endif // _LOGIT_DETAIL_COMPRESSION_UTILS_HPP_INCLUDED diff --git a/include/logit_cpp/logit/detail/MdbxByteIO.hpp b/include/logit_cpp/logit/detail/MdbxByteIO.hpp new file mode 100644 index 0000000..f5d8fc9 --- /dev/null +++ b/include/logit_cpp/logit/detail/MdbxByteIO.hpp @@ -0,0 +1,123 @@ +#pragma once + +/// \file MdbxByteIO.hpp +/// \brief Byte serialization helpers for MdbxLogger. + +#include +#include +#include +#include +#include +#include + +namespace logit { +namespace detail { + +class MdbxByteWriter { +public: + void write_u8(uint8_t value) { + m_out.push_back(value); + } + + void write_u32(uint32_t value) { + for (int shift = 24; shift >= 0; shift -= 8) { + m_out.push_back(static_cast((value >> shift) & 0xFFu)); + } + } + + void write_u64(uint64_t value) { + for (int shift = 56; shift >= 0; shift -= 8) { + m_out.push_back(static_cast((value >> shift) & 0xFFu)); + } + } + + void write_i64(int64_t value) { + uint64_t bits = 0; + std::memcpy(&bits, &value, sizeof(bits)); + write_u64(bits); + } + + void write_string(const std::string& value) { + if (value.size() > static_cast((std::numeric_limits::max)())) { + throw std::length_error("MdbxLogger: string field is too large"); + } + write_u32(static_cast(value.size())); + m_out.insert(m_out.end(), value.begin(), value.end()); + } + + const std::vector& bytes() const { + return m_out; + } + +private: + std::vector m_out; +}; + +class MdbxByteReader { +public: + MdbxByteReader(const void* data, size_t size) + : m_cur(static_cast(data)), + m_end(static_cast(data)) { + if (m_cur == nullptr && size != 0) { + throw std::runtime_error("MdbxLogger: null serialized value"); + } + m_end = m_cur == nullptr ? m_cur : m_cur + size; + } + + uint8_t read_u8() { + require(1); + return *m_cur++; + } + + uint32_t read_u32() { + require(4); + uint32_t value = 0; + for (int i = 0; i < 4; ++i) { + value = (value << 8) | static_cast(*m_cur++); + } + return value; + } + + uint64_t read_u64() { + require(8); + uint64_t value = 0; + for (int i = 0; i < 8; ++i) { + value = (value << 8) | static_cast(*m_cur++); + } + return value; + } + + int64_t read_i64() { + const uint64_t bits = read_u64(); + int64_t value = 0; + std::memcpy(&value, &bits, sizeof(value)); + return value; + } + + std::string read_string() { + const uint32_t size = read_u32(); + require(size); + const char* begin = reinterpret_cast(m_cur); + m_cur += size; + return std::string(begin, size); + } + + void finish() const { + if (m_cur != m_end) { + throw std::runtime_error("MdbxLogger: trailing bytes in serialized value"); + } + } + +private: + const uint8_t* m_cur; + const uint8_t* m_end; + + void require(size_t size) const { + if (static_cast(m_end - m_cur) < size) { + throw std::runtime_error("MdbxLogger: corrupted serialized value"); + } + } +}; + +} // namespace detail +} // namespace logit diff --git a/include/logit_cpp/logit/detail/MdbxKeyUtils.hpp b/include/logit_cpp/logit/detail/MdbxKeyUtils.hpp new file mode 100644 index 0000000..c0c0539 --- /dev/null +++ b/include/logit_cpp/logit/detail/MdbxKeyUtils.hpp @@ -0,0 +1,34 @@ +#pragma once + +/// \file MdbxKeyUtils.hpp +/// \brief Key encoding helpers for MdbxLogger record ordering. + +#include +#include +#include + +namespace logit { +namespace detail { + +inline void mdbx_write_record_key_be(std::string& key, uint64_t value, size_t offset) { + for (int shift = 56; shift >= 0; shift -= 8) { + key[offset++] = static_cast((value >> shift) & 0xFFu); + } +} + +inline void mdbx_write_record_sequence_be(std::string& key, uint32_t value, size_t offset) { + for (int shift = 24; shift >= 0; shift -= 8) { + key[offset++] = static_cast((value >> shift) & 0xFFu); + } +} + +inline std::string make_mdbx_record_key(int64_t timestamp_ms, uint32_t sequence) { + std::string key(12, '\0'); + const uint64_t sortable_ts = static_cast(timestamp_ms) ^ 0x8000000000000000ULL; + mdbx_write_record_key_be(key, sortable_ts, 0); + mdbx_write_record_sequence_be(key, sequence, 8); + return key; +} + +} // namespace detail +} // namespace logit diff --git a/include/logit_cpp/logit/detail/MdbxProcessId.hpp b/include/logit_cpp/logit/detail/MdbxProcessId.hpp new file mode 100644 index 0000000..8899809 --- /dev/null +++ b/include/logit_cpp/logit/detail/MdbxProcessId.hpp @@ -0,0 +1,26 @@ +#pragma once + +/// \file MdbxProcessId.hpp +/// \brief Cross-platform current process id helper for MdbxLogger. + +#include + +#if defined(_WIN32) +#include +#else +#include +#endif + +namespace logit { +namespace detail { + +inline uint64_t current_process_id() { +#if defined(_WIN32) + return static_cast(GetCurrentProcessId()); +#else + return static_cast(getpid()); +#endif +} + +} // namespace detail +} // namespace logit diff --git a/include/logit_cpp/logit/log_macros.hpp b/include/logit_cpp/logit/log_macros.hpp index 9a6c57c..4080976 100644 --- a/include/logit_cpp/logit/log_macros.hpp +++ b/include/logit_cpp/logit/log_macros.hpp @@ -2905,6 +2905,108 @@ static_assert(LOGIT_LEVEL_FATAL == static_cast(logit::LogLevel::LOG_LVL_FAT //------------------------------------------------------------------------------ +/// \brief Retrieves a typed backend pointer from a logger by index. +/// \param logger_index Index of logger. +/// \param logger_type Concrete logger type (e.g., logit::MdbxLogger). +/// \return Pointer to the backend, or nullptr if index/type does not match. +#define LOGIT_GET_LOGGER_AS(logger_index, logger_type) \ + (::logit::Logger::get_instance().get_logger_as((logger_index))) + +/// \brief Executes a code block when the logger backend has the requested type. +/// \param logger_index Index of logger. +/// \param logger_type Concrete logger type. +/// \param var_name Variable name available inside the block. +#define LOGIT_WITH_LOGGER_AS(logger_index, logger_type, var_name) \ + if (auto* var_name = LOGIT_GET_LOGGER_AS((logger_index), logger_type)) + +/// \brief Retrieves a read-only log-reader interface from a logger by index. +/// \param logger_index Index of logger. +/// \return Pointer to ILogReader, or nullptr if the backend does not implement it. +#define LOGIT_GET_LOG_READER(logger_index) \ + LOGIT_GET_LOGGER_AS((logger_index), ::logit::ILogReader) + +/// \brief Executes a code block when the logger backend supports ILogReader. +/// \param logger_index Index of logger. +/// \param var_name Variable name available inside the block. +#define LOGIT_WITH_LOG_READER(logger_index, var_name) \ + if (auto* var_name = LOGIT_GET_LOG_READER((logger_index))) + +/// \brief Reads a time-range of records from a backend that supports ILogReader. +/// \param logger_index Index of logger. +/// \param from_ms Inclusive start timestamp. +/// \param to_ms Exclusive end timestamp. +/// \param limit Maximum number of records (0 = unlimited). +/// \return Matching records, or empty vector if backend does not support reading. +#define LOGIT_READ_RANGE(logger_index, from_ms, to_ms, limit) \ + ([](int _idx, int64_t _from, int64_t _to, std::size_t _limit) { \ + auto* _reader = LOGIT_GET_LOG_READER(_idx); \ + return _reader \ + ? _reader->read_range(_from, _to, _limit) \ + : std::vector<::logit::LogRecordView>{}; \ + }((logger_index), (from_ms), (to_ms), (limit))) + +/// \brief Reads recent records from a backend that supports ILogReader. +/// \param logger_index Index of logger. +/// \param limit Maximum number of records (0 = unlimited). +/// \param period_ms Time window in milliseconds from now backward (0 = unlimited). +/// \param order Ascending or descending result order. +/// \return Matching records, or empty vector if backend does not support reading. +#define LOGIT_READ_RECENT(logger_index, limit, period_ms, order) \ + ([](int _idx, std::size_t _limit, int64_t _period_ms, ::logit::LogReadOrder _order) { \ + auto* _reader = LOGIT_GET_LOG_READER(_idx); \ + return _reader \ + ? _reader->read_recent(_limit, _period_ms, _order) \ + : std::vector<::logit::LogRecordView>{}; \ + }((logger_index), (limit), (period_ms), (order))) + +/// \brief Reads recent records in ascending order (oldest first). +/// \param logger_index Index of logger. +/// \param limit Maximum number of records (0 = unlimited). +/// \param period_ms Time window in milliseconds from now backward (0 = unlimited). +/// \return Matching records in ascending order, or empty vector if backend does not support reading. +#define LOGIT_READ_RECENT_ASC(logger_index, limit, period_ms) \ + LOGIT_READ_RECENT((logger_index), (limit), (period_ms), ::logit::LogReadOrder::Ascending) + +/// \brief Reads recent records in descending order (newest first). +/// \param logger_index Index of logger. +/// \param limit Maximum number of records (0 = unlimited). +/// \param period_ms Time window in milliseconds from now backward (0 = unlimited). +/// \return Matching records in descending order, or empty vector if backend does not support reading. +#define LOGIT_READ_RECENT_DESC(logger_index, limit, period_ms) \ + LOGIT_READ_RECENT((logger_index), (limit), (period_ms), ::logit::LogReadOrder::Descending) + +/// \brief Retrieves a live-subscription interface from a logger by index. +/// \param logger_index Index of logger. +/// \return Pointer to ILogSubscriber, or nullptr if the backend does not implement it. +#define LOGIT_GET_LOG_SUBSCRIBER(logger_index) \ + LOGIT_GET_LOGGER_AS((logger_index), ::logit::ILogSubscriber) + +/// \brief Executes a code block when the logger backend supports ILogSubscriber. +/// \param logger_index Index of logger. +/// \param var_name Variable name available inside the block. +#define LOGIT_WITH_LOG_SUBSCRIBER(logger_index, var_name) \ + if (auto* var_name = LOGIT_GET_LOG_SUBSCRIBER((logger_index))) + +/// \brief Registers a callback to receive newly written log records. +/// \param logger_index Index of logger. +/// \param callback Function invoked with LogRecordView after each commit. +/// \return Callback id (0 if the backend does not support subscriptions). +#define LOGIT_ADD_LOG_CALLBACK(logger_index, callback) \ + ([](int _idx, ::logit::ILogSubscriber::Callback _cb) -> uint64_t { \ + auto* _subscriber = LOGIT_GET_LOG_SUBSCRIBER(_idx); \ + return _subscriber ? _subscriber->add_log_callback(std::move(_cb)) : 0; \ + }((logger_index), (callback))) + +/// \brief Unregisters a previously added log callback. +/// \param logger_index Index of logger. +/// \param callback_id Id returned by LOGIT_ADD_LOG_CALLBACK. +/// \return True if the callback was found and removed. +#define LOGIT_REMOVE_LOG_CALLBACK(logger_index, callback_id) \ + ([](int _idx, uint64_t _id) -> bool { \ + auto* _subscriber = LOGIT_GET_LOG_SUBSCRIBER(_idx); \ + return _subscriber ? _subscriber->remove_log_callback(_id) : false; \ + }((logger_index), (callback_id))) + /// \} #endif // LOGIT_LOG_MACROS_HPP_INCLUDED diff --git a/include/logit_cpp/logit/loggers.hpp b/include/logit_cpp/logit/loggers.hpp index 5d27223..e275295 100644 --- a/include/logit_cpp/logit/loggers.hpp +++ b/include/logit_cpp/logit/loggers.hpp @@ -17,6 +17,17 @@ #include "detail/CompressionWorker.hpp" #endif +#include +#include +#include +#include +#include +#include +#include +#include + +#include "loggers/ILogReader.hpp" +#include "loggers/ILogSubscriber.hpp" #include "loggers/ILogger.hpp" #include "loggers/ConsoleLogger.hpp" #include "loggers/MemoryLogger.hpp" @@ -41,4 +52,12 @@ #include "loggers/PrometheusHttpServerLogger.hpp" #endif +#ifdef LOGIT_WITH_MDBX +#include "detail/CompressionUtils.hpp" +#include "detail/MdbxByteIO.hpp" +#include "detail/MdbxKeyUtils.hpp" +#include "detail/MdbxProcessId.hpp" +#include "loggers/MdbxLogger.hpp" +#endif + #endif // _LOGIT_LOGGERS_HPP_INCLUDED diff --git a/include/logit_cpp/logit/loggers/ILogReader.hpp b/include/logit_cpp/logit/loggers/ILogReader.hpp new file mode 100644 index 0000000..406837c --- /dev/null +++ b/include/logit_cpp/logit/loggers/ILogReader.hpp @@ -0,0 +1,70 @@ +#pragma once +#ifndef _LOGIT_ILOG_READER_HPP_INCLUDED +#define _LOGIT_ILOG_READER_HPP_INCLUDED + +/// \file ILogReader.hpp +/// \brief Optional read-only interface for log backends that support querying stored records. + +#include "../enums.hpp" +#include +#include +#include +#include + +namespace logit { + + /// \struct LogRecordView + /// \brief Common read-only view of a stored log record. + struct LogRecordView { + uint64_t session_id = 0; ///< Owning session id, or 0 if unused. + int64_t timestamp_ms = 0; ///< Log timestamp in milliseconds. + uint32_t sequence = 0; ///< Per-timestamp sequence, or 0 if unused. + LogLevel level = LogLevel::LOG_LVL_TRACE; ///< Log severity level. + std::string message; ///< Formatted message or payload preview. + uint64_t payload_id = 0; ///< Payload row id, or 0 when absent. + std::string file; ///< Source file path. + std::string function; ///< Source function name. + int line = 0; ///< Source line number. + }; + + /// \enum LogReadOrder + /// \brief Ordering for query results returned by ILogReader. + enum class LogReadOrder { + Ascending, ///< Oldest first. + Descending ///< Newest first. + }; + + /// \class ILogReader + /// \brief Optional interface for backends that expose stored records. + /// + /// Implementations should provide read_range for time-window queries + /// and read_recent for "tail -n" style access. Both return copies + /// so the caller does not depend on backend locking details. + class ILogReader { + public: + virtual ~ILogReader() = default; + + /// \brief Reads records in `[from_ms, to_ms)` ordered by timestamp. + /// \param from_ms Inclusive start timestamp. + /// \param to_ms Exclusive end timestamp. + /// \param limit Maximum number of records (0 = unlimited). + /// \return Matching records in backend-defined order (usually ascending). + virtual std::vector read_range( + int64_t from_ms, + int64_t to_ms, + std::size_t limit = 0) const = 0; + + /// \brief Reads the most recent records. + /// \param limit Maximum number of records (0 = unlimited). + /// \param period_ms Time window in milliseconds from now backward (0 = unlimited). + /// \param order Ascending (oldest first) or Descending (newest first). + /// \return Matching records in the requested order. + virtual std::vector read_recent( + std::size_t limit, + int64_t period_ms = 0, + LogReadOrder order = LogReadOrder::Ascending) const = 0; + }; + +} // namespace logit + +#endif // _LOGIT_ILOG_READER_HPP_INCLUDED diff --git a/include/logit_cpp/logit/loggers/ILogSubscriber.hpp b/include/logit_cpp/logit/loggers/ILogSubscriber.hpp new file mode 100644 index 0000000..80a7ebc --- /dev/null +++ b/include/logit_cpp/logit/loggers/ILogSubscriber.hpp @@ -0,0 +1,39 @@ +#pragma once +#ifndef _LOGIT_ILOG_SUBSCRIBER_HPP_INCLUDED +#define _LOGIT_ILOG_SUBSCRIBER_HPP_INCLUDED + +/// \file ILogSubscriber.hpp +/// \brief Optional live-subscription interface for log backends that can push newly written records. + +#include "ILogReader.hpp" +#include +#include + +namespace logit { + + /// \class ILogSubscriber + /// \brief Optional interface for backends that can notify callers when a record is successfully written. + /// + /// Callbacks receive a \ref LogRecordView containing the persisted preview and metadata. + /// They are guaranteed to be invoked only after the record has been committed to storage. + /// Implementations must not hold internal locks while invoking callbacks. + class ILogSubscriber { + public: + using Callback = std::function; + + virtual ~ILogSubscriber() = default; + + /// \brief Registers a callback to be invoked after each successfully written record. + /// \param callback Function called with the written LogRecordView. + /// \return Stable callback id that can be passed to remove_log_callback. + virtual uint64_t add_log_callback(Callback callback) = 0; + + /// \brief Unregisters a previously added callback. + /// \param callback_id Id returned by add_log_callback. + /// \return True if the callback existed and was removed. + virtual bool remove_log_callback(uint64_t callback_id) = 0; + }; + +} // namespace logit + +#endif // _LOGIT_ILOG_SUBSCRIBER_HPP_INCLUDED diff --git a/include/logit_cpp/logit/loggers/MdbxLogger.hpp b/include/logit_cpp/logit/loggers/MdbxLogger.hpp new file mode 100644 index 0000000..79b9cab --- /dev/null +++ b/include/logit_cpp/logit/loggers/MdbxLogger.hpp @@ -0,0 +1,943 @@ +#pragma once +#ifndef _LOGIT_MDBX_LOGGER_HPP_INCLUDED +#define _LOGIT_MDBX_LOGGER_HPP_INCLUDED + +/// \file MdbxLogger.hpp +/// \brief MDBX structured log storage backend. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace logit { + + /// \enum MdbxPayloadCompression + /// \brief Compression algorithm used for payload rows. + enum class MdbxPayloadCompression { + None = 0, ///< Store payload bytes as-is. + Gzip = 1, ///< Store gzip-compressed payload bytes. + Zstd = 2 ///< Store zstd-compressed payload bytes. + }; + + /// \class MdbxLogger + /// \brief Stores formatted logs in MDBX tables with optional async batching. + class MdbxLogger final : public ILogger, public ILogReader, public ILogSubscriber { + public: + /// \struct Config + /// \brief Configuration for the MDBX logger backend. + struct Config { + std::string path = "logs.mdbx"; ///< MDBX environment path. + std::string app_name; ///< Optional application name stored in session metadata. + uint64_t session_id = 0; ///< Existing session id, or 0 to auto-create. + bool async = true; ///< Queue records and write them from a worker thread. + bool drop_on_overflow = true; ///< Drop newest record when the async queue is full. + std::size_t max_queue_size = 8192; ///< Maximum async queue size; 0 means unlimited. + std::size_t max_batch_size = 256; ///< Maximum records per write transaction. + int flush_interval_ms = 100; ///< Worker wake interval. + std::size_t large_payload_threshold = 4096; ///< Spill messages larger than this. + std::size_t payload_preview_size = 512; ///< Bytes kept inline when a message spills. + bool store_large_payloads_separately = true;///< Store large messages in `log_payloads`. + MdbxPayloadCompression payload_compression = MdbxPayloadCompression::None; ///< Payload compression. + int payload_compression_level = 6; ///< Compression level for gzip/zstd. + std::function on_error; ///< Optional callback invoked on write errors instead of stderr. + }; + + /// \struct SessionView + /// \brief Public read-only view of session metadata. + struct SessionView { + std::string app_name; ///< Optional application name. + int64_t start_time_ms = 0; ///< Session start timestamp. + int64_t end_time_ms = 0; ///< Session end timestamp, or 0 while active. + uint64_t process_id = 0; ///< Process id that opened the session. + uint32_t schema_version = 1;///< Storage schema version. + }; + + /// \struct PayloadView + /// \brief Public read-only view of a large payload. + struct PayloadView { + uint64_t payload_id = 0; ///< Stable payload id. + MdbxPayloadCompression compression = MdbxPayloadCompression::None; ///< Stored compression. + std::string data; ///< Stored payload bytes. + }; + + MdbxLogger() : MdbxLogger(Config()) {} + + explicit MdbxLogger(const Config& config) + : m_config(config) { + normalize_config(); + validate_compression_config(); + open_storage(); + m_session_id = open_session(); + if (m_config.async) { + m_worker = std::thread(&MdbxLogger::worker_loop, this); + } + } + + ~MdbxLogger() override { + shutdown(); + } + + MdbxLogger(const MdbxLogger&) = delete; + MdbxLogger& operator=(const MdbxLogger&) = delete; + + /// \brief Queues or writes a formatted log record. + void log(const LogRecord& record, const std::string& message) override { + MdbxLogItem item; + item.level = record.log_level; + item.timestamp_ms = record.timestamp_ms; + item.file = record.file; + item.function = record.function; + item.line = record.line; + item.message = message; + + if (!m_config.async) { + { + std::lock_guard lock(m_mutex); + if (m_stopping) { + ++m_dropped; + return; + } + } + mark_last_log(record.timestamp_ms); + std::vector batch; + batch.push_back(std::move(item)); + write_batch(batch); + return; + } + + std::unique_lock lock(m_mutex); + if (m_stopping) { + ++m_dropped; + return; + } + + mark_last_log(record.timestamp_ms); + + if (m_config.max_queue_size > 0 && m_queue.size() >= m_config.max_queue_size) { + if (m_config.drop_on_overflow) { + ++m_dropped; + return; + } + + m_space_cv.wait(lock, [this]() { + return m_stopping || + m_config.max_queue_size == 0 || + m_queue.size() < m_config.max_queue_size; + }); + + if (m_stopping) { + ++m_dropped; + return; + } + } + + m_queue.push_back(std::move(item)); + lock.unlock(); + m_cv.notify_one(); + } + + /// \brief Waits until accepted async records are written. + void wait() override { + if (!m_config.async) { + return; + } + + std::unique_lock lock(m_mutex); + m_cv.wait(lock, [this]() { + return m_queue.empty() && m_idle; + }); + } + + /// \brief Stops the worker after draining accepted records. + void shutdown() override { + bool expected = false; + if (!m_shutdown.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) { + return; + } + + { + std::lock_guard lock(m_mutex); + m_stopping = true; + } + m_cv.notify_all(); + m_space_cv.notify_all(); + + if (m_worker.joinable()) { + m_worker.join(); + } + + try { + update_session_end(); + } catch (...) { + ++m_failed_writes; + } + } + + /// \brief Returns this logger session id. + uint64_t session_id() const { + return m_session_id; + } + + /// \brief Reads records in `[from_ms, to_ms)` ordered by timestamp and sequence. + std::vector read_range( + int64_t from_ms, + int64_t to_ms, + std::size_t limit = 0) const override { + std::vector out; + if (to_ms <= from_ms) { + return out; + } + + try { + const std::string from_key = detail::make_mdbx_record_key(from_ms, 0); + const std::string to_key = detail::make_mdbx_record_key( + to_ms - 1, + (std::numeric_limits::max)()); + + std::lock_guard db_lock(m_db_mutex); + m_records->for_each_range(from_key, to_key, + [&out, limit](const std::string&, const Record& record) -> bool { + out.push_back(to_view(record)); + return limit == 0 || out.size() < limit; + }); + } catch (...) { + out.clear(); + } + + return out; + } + + /// \brief Reads the most recent records. + /// \param limit Maximum number of records (0 = unlimited). + /// \param period_ms Time window in milliseconds from now backward (0 = unlimited). + /// \param order Ascending or descending result order. + /// \return Matching records in the requested order. + std::vector read_recent( + std::size_t limit, + int64_t period_ms = 0, + LogReadOrder order = LogReadOrder::Ascending) const override { + const int64_t now_ms = LOGIT_CURRENT_TIMESTAMP_MS(); + const int64_t from_ms = (period_ms > 0) ? (now_ms - period_ms) : 0; + auto records = read_range(from_ms, now_ms + 1, 0); + if (limit > 0 && records.size() > limit) { + records.erase( + records.begin(), + records.begin() + static_cast(records.size() - limit)); + } + if (order == LogReadOrder::Descending && !records.empty()) { + std::reverse(records.begin(), records.end()); + } + return records; + } + + /// \brief Reads a payload by id. + std::optional read_payload(uint64_t payload_id) const { + if (payload_id == 0) { + return std::nullopt; + } + + try { + std::lock_guard db_lock(m_db_mutex); +#if __cplusplus >= 201703L + auto value = m_payloads->find(payload_id); + if (!value) return std::nullopt; + return to_view(*value); +#else + std::pair value = m_payloads->find(payload_id); + if (!value.first) return std::nullopt; + return to_view(value.second); +#endif + } catch (...) { + return std::nullopt; + } + } + + /// \brief Reads and decompresses payload data by id. + /// \return Original (decompressed) payload string, or std::nullopt if not found or decompression fails. + std::optional read_payload_data(uint64_t payload_id) const { + auto view = read_payload(payload_id); + if (!view) { + return std::nullopt; + } + if (view->compression == MdbxPayloadCompression::None) { + return view->data; + } + std::string output; + bool ok = false; + if (view->compression == MdbxPayloadCompression::Gzip) { + ok = detail::decompress_string_gzip(view->data, output); + } else if (view->compression == MdbxPayloadCompression::Zstd) { + ok = detail::decompress_string_zstd(view->data, output); + } + return ok ? std::optional(std::move(output)) : std::nullopt; + } + + /// \brief Reads session metadata by id. + std::optional read_session(uint64_t session_id) const { + if (session_id == 0) { + return std::nullopt; + } + + try { + std::lock_guard db_lock(m_db_mutex); +#if __cplusplus >= 201703L + auto value = m_sessions->find(session_id); + if (!value) return std::nullopt; + return to_view(*value); +#else + std::pair value = m_sessions->find(session_id); + if (!value.first) return std::nullopt; + return to_view(value.second); +#endif + } catch (...) { + return std::nullopt; + } + } + + std::string get_string_param(const LoggerParam& param) const override { + switch (param) { + case LoggerParam::LastLogTimestamp: return std::to_string(get_last_log_ts()); + case LoggerParam::TimeSinceLastLog: return std::to_string(get_time_since_last_log()); + case LoggerParam::DroppedLogCount: return std::to_string(dropped_count()); + case LoggerParam::FailedExportCount: return std::to_string(failed_export_count()); + default: + break; + } + return std::string(); + } + + int64_t get_int_param(const LoggerParam& param) const override { + switch (param) { + case LoggerParam::LastLogTimestamp: return get_last_log_ts(); + case LoggerParam::TimeSinceLastLog: return get_time_since_last_log(); + case LoggerParam::DroppedLogCount: return counter_to_int64(dropped_count()); + case LoggerParam::FailedExportCount: return counter_to_int64(failed_export_count()); + default: + break; + } + return 0; + } + + double get_float_param(const LoggerParam& param) const override { + switch (param) { + case LoggerParam::LastLogTimestamp: + return static_cast(get_last_log_ts()) / 1000.0; + case LoggerParam::TimeSinceLastLog: + return static_cast(get_time_since_last_log()) / 1000.0; + case LoggerParam::DroppedLogCount: + return static_cast(dropped_count()); + case LoggerParam::FailedExportCount: + return static_cast(failed_export_count()); + default: + break; + } + return 0.0; + } + + void set_log_level(LogLevel level) override { + m_log_level.store(static_cast(level), std::memory_order_release); + } + + LogLevel get_log_level() const override { + return static_cast(m_log_level.load(std::memory_order_acquire)); + } + + uint64_t dropped_count() const { + return m_dropped.load(std::memory_order_acquire); + } + + uint64_t failed_export_count() const { + return m_failed_writes.load(std::memory_order_acquire); + } + + uint64_t add_log_callback(Callback callback) override { + std::lock_guard lock(m_callbacks_mutex); + const uint64_t id = m_next_callback_id.fetch_add(1, std::memory_order_relaxed); + m_callbacks.emplace(id, std::move(callback)); + return id; + } + + bool remove_log_callback(uint64_t callback_id) override { + std::lock_guard lock(m_callbacks_mutex); + return m_callbacks.erase(callback_id) > 0; + } + + private: + struct MdbxLogItem { + LogLevel level = LogLevel::LOG_LVL_TRACE; + int64_t timestamp_ms = 0; + std::string file; + std::string function; + int line = 0; + std::string message; + }; + + struct Session { + std::string app_name; + int64_t start_time_ms = 0; + int64_t end_time_ms = 0; + uint64_t process_id = 0; + uint32_t schema_version = 1; + + std::vector to_bytes() const; + static Session from_bytes(const void* data, size_t size); + }; + + struct Record { + uint64_t session_id = 0; + int64_t timestamp_ms = 0; + uint32_t sequence = 0; + LogLevel level = LogLevel::LOG_LVL_TRACE; + std::string message; + uint64_t payload_id = 0; + std::string file; + std::string function; + int line = 0; + + std::vector to_bytes() const; + static Record from_bytes(const void* data, size_t size); + }; + + struct Payload { + uint64_t payload_id = 0; + MdbxPayloadCompression compression = MdbxPayloadCompression::None; + std::string data; + + std::vector to_bytes() const; + static Payload from_bytes(const void* data, size_t size); + }; + + typedef mdbxc::KeyValueTable SessionTable; + typedef mdbxc::KeyValueTable RecordTable; + typedef mdbxc::KeyValueTable PayloadTable; + + Config m_config; + std::shared_ptr m_connection; + std::unique_ptr m_sessions; + std::unique_ptr m_records; + std::unique_ptr m_payloads; + + mutable std::mutex m_db_mutex; + + std::mutex m_mutex; + std::condition_variable m_cv; + std::condition_variable m_space_cv; + std::deque m_queue; + bool m_stopping = false; + bool m_idle = true; + std::thread m_worker; + + uint64_t m_session_id = 0; + std::unordered_map m_next_sequence_by_timestamp; + + std::atomic m_log_level = ATOMIC_VAR_INIT(static_cast(LogLevel::LOG_LVL_TRACE)); + std::atomic m_last_log_ts = ATOMIC_VAR_INIT(0); + std::atomic m_last_log_mono_ts = ATOMIC_VAR_INIT(0); + std::atomic m_dropped = ATOMIC_VAR_INIT(0); + std::atomic m_failed_writes = ATOMIC_VAR_INIT(0); + std::atomic m_shutdown = ATOMIC_VAR_INIT(false); + + mutable std::mutex m_callbacks_mutex; + std::unordered_map m_callbacks; + std::atomic m_next_callback_id{1}; + + void notify_callbacks(const std::vector& views) const { + std::vector callbacks_copy; + { + std::lock_guard lock(m_callbacks_mutex); + callbacks_copy.reserve(m_callbacks.size()); + for (const auto& kv : m_callbacks) { + callbacks_copy.push_back(kv.second); + } + } + for (const auto& view : views) { + for (const auto& cb : callbacks_copy) { + try { + cb(view); + } catch (const std::exception& e) { + if (m_config.on_error) { + m_config.on_error(std::string("MdbxLogger callback error: ") + e.what()); + } + } catch (...) { + if (m_config.on_error) { + m_config.on_error("MdbxLogger callback error"); + } + } + } + } + } + + static LogRecordView to_view(const Record& r) { + LogRecordView v; + v.session_id = r.session_id; + v.timestamp_ms = r.timestamp_ms; + v.sequence = r.sequence; + v.level = r.level; + v.message = r.message; + v.payload_id = r.payload_id; + v.file = r.file; + v.function = r.function; + v.line = r.line; + return v; + } + + static SessionView to_view(const Session& s) { + SessionView v; + v.app_name = s.app_name; + v.start_time_ms = s.start_time_ms; + v.end_time_ms = s.end_time_ms; + v.process_id = s.process_id; + v.schema_version = s.schema_version; + return v; + } + + static PayloadView to_view(const Payload& p) { + PayloadView v; + v.payload_id = p.payload_id; + v.compression = p.compression; + v.data = p.data; + return v; + } + + static std::vector serialize_session(const Session& s) { + detail::MdbxByteWriter out; + out.write_u32(1); + out.write_string(s.app_name); + out.write_i64(s.start_time_ms); + out.write_i64(s.end_time_ms); + out.write_u64(s.process_id); + out.write_u32(s.schema_version); + return out.bytes(); + } + + static Session deserialize_session(const void* data, size_t size) { + detail::MdbxByteReader in(data, size); + const uint32_t version = in.read_u32(); + if (version != 1) { + throw std::runtime_error("MdbxLogger: unsupported session value version"); + } + Session s; + s.app_name = in.read_string(); + s.start_time_ms = in.read_i64(); + s.end_time_ms = in.read_i64(); + s.process_id = in.read_u64(); + s.schema_version = in.read_u32(); + in.finish(); + return s; + } + + static std::vector serialize_record(const Record& r) { + detail::MdbxByteWriter out; + out.write_u32(1); + out.write_u64(r.session_id); + out.write_i64(r.timestamp_ms); + out.write_u32(r.sequence); + out.write_u32(static_cast(r.level)); + out.write_string(r.message); + out.write_u64(r.payload_id); + out.write_string(r.file); + out.write_string(r.function); + out.write_i64(static_cast(r.line)); + return out.bytes(); + } + + static Record deserialize_record(const void* data, size_t size) { + detail::MdbxByteReader in(data, size); + const uint32_t version = in.read_u32(); + if (version != 1) { + throw std::runtime_error("MdbxLogger: unsupported record value version"); + } + Record r; + r.session_id = in.read_u64(); + r.timestamp_ms = in.read_i64(); + r.sequence = in.read_u32(); + r.level = static_cast(in.read_u32()); + r.message = in.read_string(); + r.payload_id = in.read_u64(); + r.file = in.read_string(); + r.function = in.read_string(); + r.line = static_cast(in.read_i64()); + in.finish(); + return r; + } + + static std::vector serialize_payload(const Payload& p) { + detail::MdbxByteWriter out; + out.write_u32(1); + out.write_u64(p.payload_id); + out.write_u8(static_cast(p.compression)); + out.write_string(p.data); + return out.bytes(); + } + + static Payload deserialize_payload(const void* data, size_t size) { + detail::MdbxByteReader in(data, size); + const uint32_t version = in.read_u32(); + if (version != 1) { + throw std::runtime_error("MdbxLogger: unsupported payload value version"); + } + Payload p; + p.payload_id = in.read_u64(); + const uint8_t compression = in.read_u8(); + if (compression > static_cast(MdbxPayloadCompression::Zstd)) { + throw std::runtime_error("MdbxLogger: unsupported payload compression"); + } + p.compression = static_cast(compression); + p.data = in.read_string(); + in.finish(); + return p; + } + + void normalize_config() { + if (m_config.max_batch_size == 0) { + m_config.max_batch_size = 1; + } + if (m_config.flush_interval_ms <= 0) { + m_config.flush_interval_ms = 1; + } + } + + void validate_compression_config() const { + if (m_config.payload_compression == MdbxPayloadCompression::Gzip) { +#if !defined(LOGIT_HAS_ZLIB) + throw std::runtime_error("MdbxLogger: gzip payload compression requested but LOGIT_WITH_GZIP is not enabled"); +#endif + } + if (m_config.payload_compression == MdbxPayloadCompression::Zstd) { +#if !defined(LOGIT_HAS_ZSTD) + throw std::runtime_error("MdbxLogger: zstd payload compression requested but LOGIT_WITH_ZSTD is not enabled"); +#endif + } + } + + void open_storage() { + mdbxc::Config db_config; + db_config.pathname = m_config.path; + db_config.max_dbs = 4; + db_config.no_subdir = true; + db_config.sync_durable = true; + + m_connection = mdbxc::Connection::create(db_config); + m_sessions.reset(new SessionTable(m_connection, "log_sessions")); + m_records.reset(new RecordTable(m_connection, "log_records_by_time")); + m_payloads.reset(new PayloadTable(m_connection, "log_payloads")); + } + + uint64_t open_session() { + std::lock_guard db_lock(m_db_mutex); + auto txn = m_connection->transaction(mdbxc::TransactionMode::WRITABLE); + const int64_t now_ms = LOGIT_CURRENT_TIMESTAMP_MS(); + const uint64_t pid = detail::current_process_id(); + + if (m_config.session_id != 0) { + Session session; + if (find_session_locked(m_config.session_id, session, txn.handle())) { + if (!m_config.app_name.empty()) { + session.app_name = m_config.app_name; + } + if (session.start_time_ms == 0) { + session.start_time_ms = now_ms; + } + session.end_time_ms = 0; + session.process_id = pid; + session.schema_version = 1; + } else { + session.app_name = m_config.app_name; + session.start_time_ms = now_ms; + session.end_time_ms = 0; + session.process_id = pid; + session.schema_version = 1; + } + m_sessions->insert_or_assign(m_config.session_id, session, txn); + txn.commit(); + return m_config.session_id; + } + + for (int attempt = 0; attempt < 1024; ++attempt) { + const uint64_t candidate = make_unique_id(); + Session session; + session.app_name = m_config.app_name; + session.start_time_ms = now_ms; + session.end_time_ms = 0; + session.process_id = pid; + session.schema_version = 1; + if (m_sessions->insert(candidate, session, txn)) { + txn.commit(); + return candidate; + } + } + + throw std::runtime_error("MdbxLogger: failed to allocate unique session id"); + } + + bool find_session_locked(uint64_t session_id, Session& out, MDBX_txn* txn) const { +#if __cplusplus >= 201703L + auto value = m_sessions->find(session_id, txn); + if (!value) return false; + out = *value; + return true; +#else + std::pair value = m_sessions->find(session_id, txn); + if (!value.first) return false; + out = value.second; + return true; +#endif + } + + void update_session_end() { + std::lock_guard db_lock(m_db_mutex); + auto txn = m_connection->transaction(mdbxc::TransactionMode::WRITABLE); + Session session; + if (!find_session_locked(m_session_id, session, txn.handle())) { + session.app_name = m_config.app_name; + session.start_time_ms = LOGIT_CURRENT_TIMESTAMP_MS(); + session.process_id = detail::current_process_id(); + session.schema_version = 1; + } + session.end_time_ms = LOGIT_CURRENT_TIMESTAMP_MS(); + m_sessions->insert_or_assign(m_session_id, session, txn); + txn.commit(); + } + + void worker_loop() { + while (true) { + std::vector batch; + batch.reserve(m_config.max_batch_size); + + { + std::unique_lock lock(m_mutex); + m_cv.wait_for( + lock, + std::chrono::milliseconds(m_config.flush_interval_ms), + [this]() { return m_stopping || !m_queue.empty(); }); + + if (m_stopping && m_queue.empty()) { + m_idle = true; + m_cv.notify_all(); + return; + } + + if (m_queue.empty()) { + continue; + } + + m_idle = false; + while (!m_queue.empty() && batch.size() < m_config.max_batch_size) { + batch.push_back(std::move(m_queue.front())); + m_queue.pop_front(); + } + m_space_cv.notify_all(); + } + + write_batch(batch); + + { + std::lock_guard lock(m_mutex); + m_idle = true; + } + m_cv.notify_all(); + } + } + + void write_batch(const std::vector& batch) { + if (batch.empty()) { + return; + } + + std::vector written_views; + try { + std::lock_guard db_lock(m_db_mutex); + auto txn = m_connection->transaction(mdbxc::TransactionMode::WRITABLE); + for (size_t i = 0; i < batch.size(); ++i) { + Record record = write_item_locked(batch[i], txn); + written_views.push_back(to_view(record)); + } + txn.commit(); + } catch (const std::exception& e) { + m_failed_writes.fetch_add(1, std::memory_order_acq_rel); + if (m_config.on_error) { + m_config.on_error(std::string("MdbxLogger write error: ") + e.what()); + } + return; + } catch (...) { + m_failed_writes.fetch_add(1, std::memory_order_acq_rel); + if (m_config.on_error) { + m_config.on_error("MdbxLogger write error"); + } + return; + } + + notify_callbacks(written_views); + } + + Record write_item_locked(const MdbxLogItem& item, mdbxc::Transaction& txn) { + Record record; + record.session_id = m_session_id; + record.timestamp_ms = item.timestamp_ms; + record.level = item.level; + record.file = item.file; + record.function = item.function; + record.line = item.line; + record.message = item.message; + + if (should_spill_payload(item.message)) { + Payload payload; + payload.payload_id = allocate_payload_id_locked(txn); + fill_payload(item.message, payload); + m_payloads->insert(payload.payload_id, payload, txn); + record.payload_id = payload.payload_id; + record.message = make_payload_preview(item.message); + } + + for (;;) { + const std::string key = next_record_key_locked(record.timestamp_ms, record.sequence, txn); + if (m_records->insert(key, record, txn)) { + break; + } + advance_sequence_after_collision(record.timestamp_ms, record.sequence); + } + return record; + } + + bool should_spill_payload(const std::string& message) const { + return m_config.store_large_payloads_separately && + message.size() > m_config.large_payload_threshold; + } + + std::string make_payload_preview(const std::string& message) const { + const size_t count = std::min(message.size(), m_config.payload_preview_size); + return message.substr(0, count); + } + + void fill_payload(const std::string& message, Payload& payload) { + payload.compression = MdbxPayloadCompression::None; + payload.data = message; + + if (m_config.payload_compression == MdbxPayloadCompression::None) { + return; + } + + std::string compressed; + bool ok = false; + if (m_config.payload_compression == MdbxPayloadCompression::Gzip) { + ok = detail::compress_string_gzip(message, compressed, m_config.payload_compression_level); + } else if (m_config.payload_compression == MdbxPayloadCompression::Zstd) { + ok = detail::compress_string_zstd(message, compressed, m_config.payload_compression_level); + } + + if (ok) { + payload.compression = m_config.payload_compression; + payload.data = std::move(compressed); + } else { + m_failed_writes.fetch_add(1, std::memory_order_acq_rel); + } + } + + uint64_t allocate_payload_id_locked(mdbxc::Transaction& txn) { + for (int attempt = 0; attempt < 1024; ++attempt) { + const uint64_t candidate = make_unique_id(); + if (!m_payloads->contains(candidate, txn)) { + return candidate; + } + } + throw std::runtime_error("MdbxLogger: failed to allocate unique payload id"); + } + + std::string next_record_key_locked( + int64_t timestamp_ms, + uint32_t& sequence, + mdbxc::Transaction& txn) { + std::unordered_map::iterator it = + m_next_sequence_by_timestamp.find(timestamp_ms); + uint32_t candidate = it == m_next_sequence_by_timestamp.end() ? 0 : it->second; + + for (;;) { + const std::string key = detail::make_mdbx_record_key(timestamp_ms, candidate); + if (!m_records->contains(key, txn)) { + sequence = candidate; + if (candidate != (std::numeric_limits::max)()) { + m_next_sequence_by_timestamp[timestamp_ms] = candidate + 1; + } else { + m_next_sequence_by_timestamp[timestamp_ms] = candidate; + } + return key; + } + if (candidate == (std::numeric_limits::max)()) { + throw std::runtime_error("MdbxLogger: sequence exhausted for timestamp"); + } + ++candidate; + } + } + + void advance_sequence_after_collision(int64_t timestamp_ms, uint32_t sequence) { + if (sequence == (std::numeric_limits::max)()) { + throw std::runtime_error("MdbxLogger: sequence exhausted for timestamp"); + } + m_next_sequence_by_timestamp[timestamp_ms] = sequence + 1; + } + + static uint64_t make_unique_id() { + static std::atomic counter(0); + const uint64_t now = static_cast(LOGIT_CURRENT_TIMESTAMP_MS()); + const uint64_t n = counter.fetch_add(1, std::memory_order_relaxed) & 0xFFFFu; + const uint64_t id = (now << 16) ^ (detail::current_process_id() << 1) ^ n; + return id == 0 ? (n + 1) : id; + } + + void mark_last_log(int64_t timestamp_ms) { + m_last_log_ts.store(timestamp_ms, std::memory_order_release); + m_last_log_mono_ts.store(LOGIT_MONOTONIC_MS(), std::memory_order_release); + } + + int64_t get_last_log_ts() const { + return m_last_log_ts.load(std::memory_order_acquire); + } + + int64_t get_time_since_last_log() const { + const int64_t last_mono = m_last_log_mono_ts.load(std::memory_order_acquire); + if (last_mono <= 0) { + return 0; + } + const int64_t now = LOGIT_MONOTONIC_MS(); + return now > last_mono ? now - last_mono : 0; + } + + static int64_t counter_to_int64(uint64_t value) { + const uint64_t max_value = static_cast((std::numeric_limits::max)()); + return value > max_value ? (std::numeric_limits::max)() : static_cast(value); + } + }; + + inline std::vector MdbxLogger::Session::to_bytes() const { + return MdbxLogger::serialize_session(*this); + } + + inline MdbxLogger::Session MdbxLogger::Session::from_bytes(const void* data, size_t size) { + return MdbxLogger::deserialize_session(data, size); + } + + inline std::vector MdbxLogger::Record::to_bytes() const { + return MdbxLogger::serialize_record(*this); + } + + inline MdbxLogger::Record MdbxLogger::Record::from_bytes(const void* data, size_t size) { + return MdbxLogger::deserialize_record(data, size); + } + + inline std::vector MdbxLogger::Payload::to_bytes() const { + return MdbxLogger::serialize_payload(*this); + } + + inline MdbxLogger::Payload MdbxLogger::Payload::from_bytes(const void* data, size_t size) { + return MdbxLogger::deserialize_payload(data, size); + } + +} // namespace logit + +#endif // _LOGIT_MDBX_LOGGER_HPP_INCLUDED diff --git a/include/logit_cpp/logit/loggers/MemoryLogger.hpp b/include/logit_cpp/logit/loggers/MemoryLogger.hpp index 4355253..ef83830 100644 --- a/include/logit_cpp/logit/loggers/MemoryLogger.hpp +++ b/include/logit_cpp/logit/loggers/MemoryLogger.hpp @@ -6,14 +6,7 @@ /// \brief In-memory logger backend that stores recent log snapshots. #include "ILogger.hpp" - -#include -#include -#include -#include -#include -#include -#include +#include "ILogReader.hpp" namespace logit { @@ -23,7 +16,7 @@ namespace logit { /// \details Snapshots are returned oldest-to-newest. Read operations avoid /// `Logger`-level execution serialization, but still synchronize on this /// backend's own mutex while copying the current buffer. - class MemoryLogger : public ILogger { + class MemoryLogger : public ILogger, public ILogReader { public: /// \struct Config /// \brief Retention limits for the in-memory buffer. @@ -150,7 +143,68 @@ namespace logit { /// \brief Memory logger is synchronous, so no flush step is needed. void wait() override {} + /// \brief Reads records in `[from_ms, to_ms)` ordered by timestamp. + std::vector read_range( + int64_t from_ms, + int64_t to_ms, + std::size_t limit = 0) const override { + std::vector out; + if (to_ms <= from_ms) { + return out; + } + + std::lock_guard lock(m_mutex); + m_evict_expired_locked(LOGIT_CURRENT_TIMESTAMP_MS()); + + for (const auto& entry : m_entries) { + if (entry.timestamp_ms >= from_ms && entry.timestamp_ms < to_ms) { + out.push_back(to_view(entry)); + if (limit > 0 && out.size() >= limit) { + break; + } + } + } + return out; + } + + /// \brief Reads the most recent records. + std::vector read_recent( + std::size_t limit, + int64_t period_ms = 0, + LogReadOrder order = LogReadOrder::Ascending) const override { + std::vector out; + std::lock_guard lock(m_mutex); + m_evict_expired_locked(LOGIT_CURRENT_TIMESTAMP_MS()); + + const int64_t now_ms = LOGIT_CURRENT_TIMESTAMP_MS(); + const int64_t from_ms = (period_ms > 0) ? (now_ms - period_ms) : 0; + + for (auto it = m_entries.rbegin(); it != m_entries.rend(); ++it) { + if (period_ms <= 0 || it->timestamp_ms >= from_ms) { + out.push_back(to_view(*it)); + if (limit > 0 && out.size() >= limit) { + break; + } + } + } + + if (order == LogReadOrder::Ascending) { + std::reverse(out.begin(), out.end()); + } + return out; + } + private: + static LogRecordView to_view(const BufferedLogEntry& e) { + LogRecordView v; + v.level = e.level; + v.timestamp_ms = e.timestamp_ms; + v.file = e.file; + v.line = e.line; + v.function = e.function; + v.message = e.message; + return v; + } // Count only the retained formatted payload, not the full object footprint. static std::size_t m_entry_bytes(const BufferedLogEntry& entry) { return entry.message.size(); diff --git a/include/logit_cpp/logit/loggers/otlp/OtlpCompression.hpp b/include/logit_cpp/logit/loggers/otlp/OtlpCompression.hpp index 3b83df7..c92debb 100644 --- a/include/logit_cpp/logit/loggers/otlp/OtlpCompression.hpp +++ b/include/logit_cpp/logit/loggers/otlp/OtlpCompression.hpp @@ -2,106 +2,21 @@ #ifndef _LOGIT_OTLP_COMPRESSION_HPP_INCLUDED #define _LOGIT_OTLP_COMPRESSION_HPP_INCLUDED -#include -#include +/// \file OtlpCompression.hpp +/// \brief Backward-compatible forwarding header for shared compression helpers. +/// +/// New code should include `` directly. +/// This header is kept for existing OTLP headers that include it. -#if defined(LOGIT_HAS_ZLIB) -# include -#endif - -#if defined(LOGIT_HAS_ZSTD) -# include -#endif +#include "../../detail/CompressionUtils.hpp" namespace logit { enum class OtlpCompression { None, Gzip, Zstd }; -/// \brief Compress a string with gzip. -/// \param input Uncompressed data. -/// \param[out] output Compressed result (valid only on success). -/// \param level Compression level 1-9. -/// \return true on success, false if zlib is unavailable or compression fails. -/// \note Callers must check the return value and handle fallback explicitly. -inline bool compress_string_gzip(const std::string& input, std::string& output, int level) { -#if defined(LOGIT_HAS_ZLIB) - z_stream zs; - zs.zalloc = Z_NULL; - zs.zfree = Z_NULL; - zs.opaque = Z_NULL; - zs.next_in = reinterpret_cast(const_cast(input.data())); - zs.avail_in = static_cast(input.size()); - zs.next_out = Z_NULL; - zs.avail_out = 0; - - if (level < 1) level = 1; - if (level > 9) level = 9; - - int window_bits = 15 + 16; - if (deflateInit2(&zs, level, Z_DEFLATED, window_bits, 8, Z_DEFAULT_STRATEGY) != Z_OK) { - return false; - } - - output.clear(); - - int ret = Z_OK; - std::size_t offset = 0; - const std::size_t buf_size = 32768; - - do { - output.resize(offset + buf_size); - zs.next_out = reinterpret_cast(&output[offset]); - zs.avail_out = static_cast(buf_size); - - ret = deflate(&zs, Z_FINISH); - if (ret == Z_STREAM_ERROR) { - deflateEnd(&zs); - return false; - } - - offset = zs.total_out; - } while (ret != Z_STREAM_END); - - output.resize(zs.total_out); - deflateEnd(&zs); - return true; -#else - (void)input; (void)output; (void)level; - return false; -#endif -} - -/// \brief Compress a string with zstd. -/// \param input Uncompressed data. -/// \param[out] output Compressed result (valid only on success). -/// \param level Compression level 1-19. -/// \return true on success, false if zstd is unavailable or compression fails. -/// \note Callers must check the return value and handle fallback explicitly. -inline bool compress_string_zstd(const std::string& input, std::string& output, int level) { -#if defined(LOGIT_HAS_ZSTD) - if (level < 1) level = 1; - if (level > 19) level = 19; - - std::size_t bound = ZSTD_compressBound(input.size()); - output.resize(bound); - - std::size_t result = ZSTD_compress( - &output[0], output.size(), - input.data(), input.size(), - level); - - if (ZSTD_isError(result)) { - output.clear(); - return false; - } - - output.resize(result); - return true; -#else - (void)input; (void)output; (void)level; - return false; -#endif -} +// Forward declarations: the actual implementations live in detail:: +using detail::compress_string_gzip; +using detail::compress_string_zstd; } // namespace logit diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index f1a9dad..073947a 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -48,6 +48,7 @@ else() memory_logger_backend_test.cpp memory_logger_concurrency_test.cpp memory_logger_integration_test.cpp + mdbx_logger_test.cpp mdc_ndc_context_test.cpp os_error_macros_test.cpp otlp_http_logger_integration_test.cpp @@ -109,6 +110,9 @@ else() list(REMOVE_ITEM TEST_SOURCES prometheus_metric_builders_test.cpp) list(REMOVE_ITEM TEST_SOURCES prometheus_registry_test.cpp) endif() + if(NOT LOGIT_WITH_MDBX) + list(REMOVE_ITEM TEST_SOURCES mdbx_logger_test.cpp) + endif() if(LOGIT_WITH_PROMETHEUS AND NOT LOGIT_WITH_PROMETHEUS_SERVER) list(REMOVE_ITEM TEST_SOURCES prometheus_http_server_logger_test.cpp) endif() diff --git a/tests/mdbx_logger_test.cpp b/tests/mdbx_logger_test.cpp new file mode 100644 index 0000000..c37dd90 --- /dev/null +++ b/tests/mdbx_logger_test.cpp @@ -0,0 +1,430 @@ +#include +#include + +#ifdef LOGIT_WITH_MDBX + +#include +#include +#include +#include +#include +#include + +namespace { + +std::string make_db_path(const std::string& suffix) { + std::ostringstream os; + os << logit::get_exec_dir() + << "/mdbx_logger_test_" + << suffix + << "_" + << LOGIT_CURRENT_TIMESTAMP_MS() + << ".mdbx"; + return os.str(); +} + +void cleanup_db(const std::string& path) { + std::remove(path.c_str()); + std::remove((path + "-lck").c_str()); +} + +logit::LogRecord make_record(logit::LogLevel level, int64_t timestamp_ms, int line) { + return logit::LogRecord( + level, + timestamp_ms, + "mdbx_logger_test.cpp", + line, + "make_record", + "message", + "", + -1, + false, + false, + false); +} + +void test_sync_range_session_and_sequences() { + const std::string path = make_db_path("sync"); + cleanup_db(path); + + { + logit::MdbxLogger::Config config; + config.path = path; + config.app_name = "mdbx-test"; + config.async = false; + + logit::MdbxLogger logger(config); + const uint64_t session_id = logger.session_id(); + assert(session_id != 0); + + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 2001, 10), "later-1"); + logger.log(make_record(logit::LogLevel::LOG_LVL_WARN, 2000, 11), "backward"); + logger.log(make_record(logit::LogLevel::LOG_LVL_ERROR, 2001, 12), "later-2"); + + auto records = logger.read_range(2000, 2002); + assert(records.size() == 3); + assert(records[0].timestamp_ms == 2000); + assert(records[0].message == "backward"); + assert(records[1].timestamp_ms == 2001); + assert(records[2].timestamp_ms == 2001); + assert(records[1].sequence != records[2].sequence); + assert(records[2].sequence > records[1].sequence); + assert(records[2].file == "mdbx_logger_test.cpp"); + assert(records[2].function == "make_record"); + assert(records[2].line == 12); + assert(records[2].session_id == session_id); + + auto limited = logger.read_range(2000, 2002, 2); + assert(limited.size() == 2); + + auto session_opt = logger.read_session(session_id); + assert(session_opt); + assert(session_opt->app_name == "mdbx-test"); + assert(session_opt->start_time_ms > 0); + assert(session_opt->end_time_ms == 0); + assert(session_opt->schema_version == 1); + + logger.shutdown(); + session_opt = logger.read_session(session_id); + assert(session_opt); + assert(session_opt->end_time_ms >= session_opt->start_time_ms); + } + + cleanup_db(path); +} + +void test_async_large_payload_spill() { + const std::string path = make_db_path("payload"); + cleanup_db(path); + + { + logit::MdbxLogger::Config config; + config.path = path; + config.async = true; + config.flush_interval_ms = 5; + config.max_batch_size = 8; + config.large_payload_threshold = 10; + config.payload_preview_size = 5; + config.store_large_payloads_separately = true; + + logit::MdbxLogger logger(config); + + const std::string small = "small"; + const std::string large = "abcdefghijklmnopqrstuvwxyz"; + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 3000, 20), small); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 3001, 21), large); + logger.wait(); + + auto records = logger.read_range(3000, 3002); + assert(records.size() == 2); + assert(records[0].message == small); + assert(records[0].payload_id == 0); + assert(records[1].message == "abcde"); + assert(records[1].payload_id != 0); + + auto payload_opt = logger.read_payload(records[1].payload_id); + assert(payload_opt); + assert(payload_opt->payload_id == records[1].payload_id); + assert(payload_opt->compression == logit::MdbxPayloadCompression::None); + assert(payload_opt->data == large); + + auto data_opt = logger.read_payload_data(records[1].payload_id); + assert(data_opt); + assert(*data_opt == large); + + logger.shutdown(); + } + + cleanup_db(path); +} + +#if defined(LOGIT_HAS_ZLIB) +void test_gzip_payload_compression() { + const std::string path = make_db_path("gzip"); + cleanup_db(path); + + { + logit::MdbxLogger::Config config; + config.path = path; + config.async = false; + config.large_payload_threshold = 4; + config.payload_preview_size = 3; + config.payload_compression = logit::MdbxPayloadCompression::Gzip; + config.payload_compression_level = 6; + + logit::MdbxLogger logger(config); + const std::string large(128, 'x'); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 4000, 30), large); + + auto records = logger.read_range(4000, 4001); + assert(records.size() == 1); + assert(records[0].payload_id != 0); + + auto payload_opt = logger.read_payload(records[0].payload_id); + assert(payload_opt); + assert(payload_opt->compression == logit::MdbxPayloadCompression::Gzip); + assert(!payload_opt->data.empty()); + assert(payload_opt->data != large); + + auto data_opt = logger.read_payload_data(records[0].payload_id); + assert(data_opt); + assert(*data_opt == large); + + logger.shutdown(); + } + + cleanup_db(path); +} +#endif + +void test_counters_zero_for_sync_writes() { + const std::string path = make_db_path("counters"); + cleanup_db(path); + + { + logit::MdbxLogger::Config config; + config.path = path; + config.async = false; + + logit::MdbxLogger logger(config); + assert(logger.dropped_count() == 0); + assert(logger.failed_export_count() == 0); + + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 5000, 40), "first"); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 5001, 41), "second"); + + assert(logger.dropped_count() == 0); + assert(logger.failed_export_count() == 0); + logger.shutdown(); + } + + cleanup_db(path); +} + +void test_on_error_callback() { + const std::string path = make_db_path("error_cb"); + cleanup_db(path); + + { + std::vector errors; + logit::MdbxLogger::Config config; + config.path = path; + config.async = false; + config.on_error = [&errors](const std::string& msg) { + errors.push_back(msg); + }; + + logit::MdbxLogger logger(config); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 6000, 50), "ok"); + logger.shutdown(); + assert(errors.empty()); + } + + cleanup_db(path); +} + +void test_read_range_empty_and_limits() { + const std::string path = make_db_path("range"); + cleanup_db(path); + + { + logit::MdbxLogger::Config config; + config.path = path; + config.async = false; + + logit::MdbxLogger logger(config); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 7000, 60), "a"); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 7001, 61), "b"); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 7002, 62), "c"); + + auto empty = logger.read_range(8000, 9000); + assert(empty.empty()); + + auto all = logger.read_range(7000, 7003); + assert(all.size() == 3); + + auto limited = logger.read_range(7000, 7003, 2); + assert(limited.size() == 2); + + auto zero_span = logger.read_range(7001, 7001); + assert(zero_span.empty()); + + logger.shutdown(); + } + + cleanup_db(path); +} + +void test_read_recent() { + const std::string path = make_db_path("recent"); + cleanup_db(path); + + { + logit::MdbxLogger::Config config; + config.path = path; + config.async = false; + + logit::MdbxLogger logger(config); + const int64_t now = LOGIT_CURRENT_TIMESTAMP_MS(); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, now - 3000, 70), "old"); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, now - 1000, 71), "mid"); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, now, 72), "new"); + + auto all_asc = logger.read_recent(0, 0, logit::LogReadOrder::Ascending); + assert(all_asc.size() == 3); + assert(all_asc[0].message == "old"); + assert(all_asc[2].message == "new"); + + auto all_desc = logger.read_recent(0, 0, logit::LogReadOrder::Descending); + assert(all_desc.size() == 3); + assert(all_desc[0].message == "new"); + assert(all_desc[2].message == "old"); + + auto limited = logger.read_recent(2, 0, logit::LogReadOrder::Ascending); + assert(limited.size() == 2); + assert(limited[0].message == "mid"); + assert(limited[1].message == "new"); + + auto limited_desc = logger.read_recent(2, 0, logit::LogReadOrder::Descending); + assert(limited_desc.size() == 2); + assert(limited_desc[0].message == "new"); + assert(limited_desc[1].message == "mid"); + + auto period = logger.read_recent(10, 1500, logit::LogReadOrder::Ascending); + assert(period.size() == 2); + assert(period[0].message == "mid"); + assert(period[1].message == "new"); + + logger.shutdown(); + } + + cleanup_db(path); +} + +void test_callback_sync() { + const std::string path = make_db_path("cb_sync"); + cleanup_db(path); + + { + logit::MdbxLogger::Config config; + config.path = path; + config.async = false; + + logit::MdbxLogger logger(config); + std::vector received; + const uint64_t cb_id = logger.add_log_callback( + [&received](const logit::LogRecordView& v) { + received.push_back(v); + }); + assert(cb_id != 0); + + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 8000, 80), "hello"); + assert(received.size() == 1); + assert(received[0].message == "hello"); + assert(received[0].level == logit::LogLevel::LOG_LVL_INFO); + + assert(logger.remove_log_callback(cb_id)); + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 8001, 81), "after-remove"); + assert(received.size() == 1); + + logger.shutdown(); + } + + cleanup_db(path); +} + +void test_callback_async() { + const std::string path = make_db_path("cb_async"); + cleanup_db(path); + + { + logit::MdbxLogger::Config config; + config.path = path; + config.async = true; + config.flush_interval_ms = 5; + config.max_batch_size = 64; + + logit::MdbxLogger logger(config); + std::vector received; + const uint64_t cb_id = logger.add_log_callback( + [&received](const logit::LogRecordView& v) { + received.push_back(v); + }); + assert(cb_id != 0); + + logger.log(make_record(logit::LogLevel::LOG_LVL_WARN, 9000, 90), "async-1"); + logger.log(make_record(logit::LogLevel::LOG_LVL_ERROR, 9001, 91), "async-2"); + logger.wait(); + + assert(received.size() == 2); + assert(received[0].message == "async-1"); + assert(received[1].message == "async-2"); + + assert(logger.remove_log_callback(cb_id)); + logger.shutdown(); + } + + cleanup_db(path); +} + +void test_callback_exception_safe() { + const std::string path = make_db_path("cb_ex"); + cleanup_db(path); + + { + logit::MdbxLogger::Config config; + config.path = path; + config.async = false; + std::vector errors; + config.on_error = [&errors](const std::string& msg) { + errors.push_back(msg); + }; + + logit::MdbxLogger logger(config); + bool second_called = false; + + logger.add_log_callback( + [](const logit::LogRecordView&) { + throw std::runtime_error("boom"); + }); + logger.add_log_callback( + [&second_called](const logit::LogRecordView&) { + second_called = true; + }); + + logger.log(make_record(logit::LogLevel::LOG_LVL_INFO, 10000, 100), "ok"); + assert(second_called); + assert(!errors.empty()); + + logger.shutdown(); + } + + cleanup_db(path); +} + +} // namespace + +int main() { + test_sync_range_session_and_sequences(); + test_async_large_payload_spill(); +#if defined(LOGIT_HAS_ZLIB) + test_gzip_payload_compression(); +#endif + test_counters_zero_for_sync_writes(); + test_on_error_callback(); + test_read_range_empty_and_limits(); + test_read_recent(); + test_callback_sync(); + test_callback_async(); + test_callback_exception_safe(); + std::cout << "PASS: mdbx_logger_test" << std::endl; + return 0; +} + +#else + +int main() { + return 0; +} + +#endif