diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8ad59c8..2194eed 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -182,6 +182,12 @@ jobs: run: cmake --build build-otlp --parallel 2 - name: Run OTLP tests run: ctest --test-dir build-otlp --output-on-failure -R '^otlp_' + - name: Run compressed file reader tests + run: >- + ctest --test-dir build-otlp + --output-on-failure + --timeout 10 + -R '^file_logger_(gzip_compression|truncated_gzip|zstd_compression)_test$' - name: Upload OTLP logs if: failure() uses: actions/upload-artifact@v4 diff --git a/bench/adapters/LogItAdapter.cpp b/bench/adapters/LogItAdapter.cpp index 1a47b17..e7bb037 100644 --- a/bench/adapters/LogItAdapter.cpp +++ b/bench/adapters/LogItAdapter.cpp @@ -1,6 +1,7 @@ #include "LogItAdapter.hpp" #include +#include #include #include #include @@ -89,9 +90,19 @@ namespace logit_bench { void wait() override { if (m_async) { - logit::detail::TaskExecutor::get_instance().wait(); + const std::uint64_t generation = flush_generation(); + logit::detail::TaskExecutor::get_instance().add_task([this]() { + std::lock_guard lock(m_flush_mutex); + ++m_flush_generation; + m_flush_cv.notify_all(); + }); + + std::unique_lock lock(m_flush_mutex); + m_flush_cv.wait(lock, [this, generation]() { + return m_flush_generation > generation; + }); } - + std::lock_guard lock(m_file_mutex); if (m_file.is_open()) { m_file.flush(); @@ -118,6 +129,11 @@ namespace logit_bench { } } } + + std::uint64_t flush_generation() const { + std::lock_guard lock(m_flush_mutex); + return m_flush_generation; + } bool m_async = false; SinkKind m_sink = SinkKind::Null; @@ -125,7 +141,11 @@ namespace logit_bench { std::ofstream m_file; mutable std::mutex m_file_mutex; - + + mutable std::mutex m_flush_mutex; + std::condition_variable m_flush_cv; + std::uint64_t m_flush_generation = 0; + std::atomic m_level{static_cast(logit::LogLevel::LOG_LVL_TRACE)}; }; diff --git a/include/logit_cpp/logit/detail/CompressionUtils.hpp b/include/logit_cpp/logit/detail/CompressionUtils.hpp index 308cb2a..fc19d17 100644 --- a/include/logit_cpp/logit/detail/CompressionUtils.hpp +++ b/include/logit_cpp/logit/detail/CompressionUtils.hpp @@ -103,14 +103,28 @@ inline bool decompress_string_gzip(const std::string& input, std::string& output zs.next_out = reinterpret_cast(&output[offset]); zs.avail_out = static_cast(buf_size); + const uInt previous_avail_in = zs.avail_in; + const uLong previous_total_out = zs.total_out; ret = inflate(&zs, Z_NO_FLUSH); - if (ret == Z_STREAM_ERROR || ret == Z_DATA_ERROR || ret == Z_MEM_ERROR) { + + // Only these two results are valid for the loop below. In particular, + // Z_BUF_ERROR means that inflate made no progress (typically because a + // compressed stream was truncated after consuming all input). + if (ret != Z_OK && ret != Z_STREAM_END) { inflateEnd(&zs); output.clear(); return false; } offset = zs.total_out; + + if (ret == Z_OK && + zs.avail_in == previous_avail_in && + zs.total_out == previous_total_out) { + inflateEnd(&zs); + output.clear(); + return false; + } } while (ret != Z_STREAM_END); output.resize(zs.total_out); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 3492593..7d9713b 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -25,6 +25,7 @@ else() file_logger_external_cmd_compression_test.cpp file_logger_file_api_test.cpp file_logger_gzip_compression_test.cpp + file_logger_truncated_gzip_test.cpp file_logger_remove_old_logs_suffixes_test.cpp file_logger_rotation_naming_sequence_test.cpp file_logger_rotation_naming_timestamp_ms_test.cpp @@ -89,6 +90,7 @@ else() ) if(NOT LOGIT_WITH_GZIP) list(REMOVE_ITEM TEST_SOURCES file_logger_gzip_compression_test.cpp) + list(REMOVE_ITEM TEST_SOURCES file_logger_truncated_gzip_test.cpp) list(REMOVE_ITEM TEST_SOURCES file_logger_external_cmd_compression_test.cpp) list(REMOVE_ITEM TEST_SOURCES otlp_http_logger_gzip_test.cpp) endif() diff --git a/tests/file_logger_truncated_gzip_test.cpp b/tests/file_logger_truncated_gzip_test.cpp new file mode 100644 index 0000000..21b2b56 --- /dev/null +++ b/tests/file_logger_truncated_gzip_test.cpp @@ -0,0 +1,109 @@ +#include + +#if defined(LOGIT_HAS_ZLIB) + +#include + +#include +#include +#include +#include +#include + +namespace { + +std::string make_unique_directory_name() { + const long long stamp = static_cast( + std::chrono::steady_clock::now().time_since_epoch().count()); + return "truncated_gzip_logs_" + std::to_string(stamp); +} + +std::string make_rotated_path(const std::string& current_path) { + std::string rotated = current_path; + const size_t pos = rotated.rfind(".log"); + if (pos != std::string::npos) { + rotated.insert(pos, ".001"); + } + return rotated + ".gz"; +} + +bool write_gzip_file(const std::string& path, const std::string& content) { + gzFile file = gzopen(path.c_str(), "wb"); + if (!file) return false; + + const int written = gzwrite( + file, + content.data(), + static_cast(content.size())); + const int close_result = gzclose(file); + return written == static_cast(content.size()) && close_result == Z_OK; +} + +bool read_binary_file(const std::string& path, std::string& content) { + std::ifstream file(path.c_str(), std::ios_base::binary); + if (!file.is_open()) return false; + + std::ostringstream stream; + stream << file.rdbuf(); + content = stream.str(); + return !file.bad(); +} + +bool write_binary_file(const std::string& path, const std::string& content) { + std::ofstream file(path.c_str(), std::ios_base::binary | std::ios_base::trunc); + if (!file.is_open()) return false; + + file.write(content.data(), static_cast(content.size())); + return file.good(); +} + +} // namespace + +int main() { + const std::string directory = make_unique_directory_name(); + const std::string payload = "truncated gzip payload\n"; + std::string current_path; + std::string compressed_path; + int result_code = 1; + + { + logit::FileLogger::Config config; + config.directory = directory; + config.async = false; + + logit::FileLogger logger(config); + current_path = logger.get_string_param(logit::LoggerParam::LastFilePath); + compressed_path = make_rotated_path(current_path); + + if (!write_gzip_file(compressed_path, payload)) { + logger.shutdown(); + } else { + const logit::LogFileReadResult valid = + logger.read_log_file(compressed_path); + + std::string compressed_bytes; + if (valid.ok && valid.content == payload && + read_binary_file(compressed_path, compressed_bytes) && + compressed_bytes.size() > 8) { + compressed_bytes.resize(compressed_bytes.size() - 8); + if (write_binary_file(compressed_path, compressed_bytes)) { + const logit::LogFileReadResult truncated = + logger.read_log_file(compressed_path); + result_code = + !truncated.ok && truncated.content.empty() ? 0 : 1; + } + } + logger.shutdown(); + } + } + + std::remove(compressed_path.c_str()); + std::remove(current_path.c_str()); + return result_code; +} + +#else + +int main() { return 0; } + +#endif