diff --git a/components/core/cmake/Options/options.cmake b/components/core/cmake/Options/options.cmake index 1c8c327360..3802b3dde4 100644 --- a/components/core/cmake/Options/options.cmake +++ b/components/core/cmake/Options/options.cmake @@ -270,6 +270,7 @@ function(set_clp_s_io_dependencies) set_clp_need_flags( CLP_NEED_BOOST CLP_NEED_FMT + CLP_NEED_SIMDJSON CLP_NEED_SPDLOG CLP_NEED_ZSTD ) diff --git a/components/core/src/clp_s/CMakeLists.txt b/components/core/src/clp_s/CMakeLists.txt index b768488565..bb45ea8f4f 100644 --- a/components/core/src/clp_s/CMakeLists.txt +++ b/components/core/src/clp_s/CMakeLists.txt @@ -214,6 +214,7 @@ if(CLP_BUILD_CLP_S_IO) Boost::iostreams Boost::url clp_s::clp_dependencies fmt::fmt + simdjson::simdjson spdlog::spdlog ${zstd_TARGET} ) diff --git a/components/core/src/clp_s/InputConfig.cpp b/components/core/src/clp_s/InputConfig.cpp index e94c8bb8db..d4c892a14f 100644 --- a/components/core/src/clp_s/InputConfig.cpp +++ b/components/core/src/clp_s/InputConfig.cpp @@ -7,9 +7,11 @@ #include #include #include +#include #include #include +#include #include #include "../clp/aws/AwsAuthenticationSigner.hpp" @@ -21,6 +23,7 @@ #include "../clp/spdlog_with_specializations.hpp" #include "../clp/streaming_compression/Decompressor.hpp" #include "../clp/streaming_compression/zstd/Decompressor.hpp" +#include "../clp/utf8_utils.hpp" #include "Utils.hpp" namespace clp_s { @@ -134,6 +137,14 @@ auto could_be_kvir(char const* peek_buf, size_t peek_size) -> bool; */ auto could_be_json(char const* peek_buf, size_t peek_size) -> bool; +/** + * Checks if an input contains logtext, based on the first few bytes of data from the input. + * @param peek_buf A pointer to a buffer containing peeked data from the start of an input stream. + * @param peek_size The number of bytes of peeked data in the buffer. + * @return Whether the input could be logtext. + */ +auto could_be_logtext(char const* peek_buf, size_t peek_size) -> bool; + auto try_create_file_reader(std::string_view const file_path) -> std::shared_ptr { try { @@ -242,6 +253,65 @@ auto could_be_json(char const* peek_buf, size_t peek_size) -> bool { return false; } +/** + * This function decides whether the buffer could be logtext based on whether it contains valid, but + * potentially truncated, UTF-8 data. + * + * To check for valid UTF-8 while accounting for truncation we need to: + * 1. Find the last complete UTF-8 codepoint in the buffer + * 2. Validate that all of the data in the buffer including the last complete codepoint is valid + * UTF-8. + * + * For the first step we can always find the last complete codepoint by scanning the last 7 bytes of + * the buffer. This is because in the worst case, the stream terminates with a 4-byte codepoint + * followed by another 4-byte codepoint with its last byte truncated. The approach, then, is to + * scan the last 7 bytes of the buffer to locate the last complete codepoint. Since we use full + * UTF-8 validation in the second step, this scan only needs to look for UTF-8 header bytes without + * validating continuation bytes. If no complete codepoint can be found, we know that the input is + * not valid UTF-8, otherwise we continue on to validating the entire buffer through the end of the + * last complete codepoint. + * + * For the second step, we simply use an off-the-shelf fast UTF-8 validator. + */ +auto could_be_logtext(char const* peek_buf, size_t peek_size) -> bool { + constexpr size_t cMaxUtf8CodepointBytes = 4ULL; + constexpr size_t cMaxRunWithoutFullUtf8Codepoint = 2 * cMaxUtf8CodepointBytes - 1ULL; + + size_t cur_byte{ + peek_size < cMaxRunWithoutFullUtf8Codepoint + ? 0ULL + : (peek_size - cMaxRunWithoutFullUtf8Codepoint) + }; + + std::optional legal_last_byte_index{std::nullopt}; + auto mark_last_legal_character = [&](size_t remaining_bytes_in_char) { + auto const last_byte_in_char = cur_byte + remaining_bytes_in_char; + if (last_byte_in_char < peek_size) { + legal_last_byte_index = last_byte_in_char; + } + cur_byte = last_byte_in_char; + }; + + for (; cur_byte < peek_size; ++cur_byte) { + uint8_t const c{static_cast(peek_buf[cur_byte])}; + if ((clp::cFourByteUtf8CharHeaderMask & c) == clp::cFourByteUtf8CharHeader) { + mark_last_legal_character(3ULL); + } else if ((clp::cThreeByteUtf8CharHeaderMask & c) == clp::cThreeByteUtf8CharHeader) { + mark_last_legal_character(2ULL); + } else if ((clp::cTwoByteUtf8CharHeaderMask & c) == clp::cTwoByteUtf8CharHeader) { + mark_last_legal_character(1ULL); + } else if (clp::utf8_utils_internal::is_ascii_char(c)) { + mark_last_legal_character(0ULL); + } + } + + if (false == legal_last_byte_index.has_value()) { + return false; + } + + return simdjson::validate_utf8(peek_buf, legal_last_byte_index.value() + 1ULL); +} + auto peek_start_and_deduce_type(std::shared_ptr& reader) -> FileType { char const* peek_buf{}; size_t peek_size{}; @@ -262,6 +332,10 @@ auto peek_start_and_deduce_type(std::shared_ptr& reader) -> return FileType::Json; } + if (could_be_logtext(peek_buf, peek_size)) { + return FileType::LogText; + } + return FileType::Unknown; } } // namespace @@ -307,6 +381,7 @@ auto try_create_reader(Path const& path, NetworkAuthOption const& network_auth) switch (type) { case FileType::Json: case FileType::KeyValueIr: + case FileType::LogText: return {std::move(readers), type}; case FileType::Zstd: { readers.emplace_back( diff --git a/components/core/src/clp_s/InputConfig.hpp b/components/core/src/clp_s/InputConfig.hpp index 45c880f06e..15562d8807 100644 --- a/components/core/src/clp_s/InputConfig.hpp +++ b/components/core/src/clp_s/InputConfig.hpp @@ -23,6 +23,7 @@ constexpr char cAwsSessionTokenEnvVar[] = "AWS_SESSION_TOKEN"; enum class FileType : uint8_t { Json = 0, KeyValueIr, + LogText, Zstd, Unknown }; diff --git a/components/core/src/clp_s/JsonParser.cpp b/components/core/src/clp_s/JsonParser.cpp index 2d45581842..8fbf597b18 100644 --- a/components/core/src/clp_s/JsonParser.cpp +++ b/components/core/src/clp_s/JsonParser.cpp @@ -637,6 +637,13 @@ bool JsonParser::ingest() { case FileType::KeyValueIr: ingestion_successful = ingest_kvir(nested_readers.back(), path, archive_creator_id); break; + case FileType::LogText: + SPDLOG_ERROR( + "Direct ingestion of unstructured logtext is not supported from input {}", + path.path + ); + std::ignore = m_archive_writer->close(); + return false; case FileType::Zstd: case FileType::Unknown: default: {