Merge branch 'main' into helm-admin

Author: junhaoliao

components/core/src/clp_s/ArchiveReader.cpp

@@ -30,7 +30,29 @@ void ArchiveReader::open(Path const& archive_path, NetworkAuthOption const& netw
     }
 
     m_archive_reader_adaptor = std::make_shared<ArchiveReaderAdaptor>(archive_path, network_auth);
+    initialize_archive_reader();
+}
+
+auto ArchiveReader::open(
+        std::shared_ptr<clp::ReaderInterface> single_file_archive_reader,
+        std::string_view archive_id
+) -> void {
+    if (m_is_open) {
+        throw OperationFailed(ErrorCodeNotReady, __FILENAME__, __LINE__);
+    }
+    m_is_open = true;
+
+    if (nullptr == single_file_archive_reader || archive_id.empty()) {
+        throw OperationFailed(ErrorCodeBadParam, __FILENAME__, __LINE__);
+    }
+    m_archive_id = archive_id;
+
+    m_archive_reader_adaptor
+            = std::make_shared<ArchiveReaderAdaptor>(std::move(single_file_archive_reader));
+    initialize_archive_reader();
+}
 
+auto ArchiveReader::initialize_archive_reader() -> void {
     if (auto const rc = m_archive_reader_adaptor->load_archive_metadata(); ErrorCodeSuccess != rc) {
         throw OperationFailed(rc, __FILENAME__, __LINE__);
     }

components/core/src/clp_s/ArchiveReader.hpp

@@ -41,6 +41,16 @@ class ArchiveReader {
      */
     void open(Path const& archive_path, NetworkAuthOption const& network_auth);
 
+    /**
+     * Opens a single-file archive for reading from an already open `clp::ReaderInterface`.
+     * @param single_file_archive_reader The already opened archive reader
+     * @param archive_id The unique name or identifier for the archive
+     */
+    auto open(
+            std::shared_ptr<clp::ReaderInterface> single_file_archive_reader,
+            std::string_view archive_id
+    ) -> void;
+
     /**
      * Reads the dictionaries and metadata.
      */
@@ -164,6 +174,11 @@ class ArchiveReader {
     }
 
 private:
+    /**
+     * Reads archive metadata and prepares the archive reader for subsequent archive reads.
+     */
+    auto initialize_archive_reader() -> void;
+
     /**
      * Initializes a schema reader passed by reference to become a reader for a given schema.
      * @param reader

components/core/src/clp_s/ArchiveReaderAdaptor.cpp

@@ -14,8 +14,10 @@
 #include <spdlog/spdlog.h>
 
 #include "../clp/BoundedReader.hpp"
+#include "../clp/ErrorCode.hpp"
 #include "../clp/FileReader.hpp"
 #include "archive_constants.hpp"
+#include "ErrorCode.hpp"
 #include "InputConfig.hpp"
 #include "RangeIndexWriter.hpp"
 #include "SingleFileArchiveDefs.hpp"
@@ -27,7 +29,6 @@ ArchiveReaderAdaptor::ArchiveReaderAdaptor(
 )
         : m_archive_path{archive_path},
           m_network_auth{network_auth},
-          m_single_file_archive{false},
           m_timestamp_dictionary{std::make_shared<TimestampDictionaryReader>()} {
     if (InputSource::Filesystem != archive_path.source
         || std::filesystem::is_regular_file(archive_path.path))
@@ -36,6 +37,21 @@ ArchiveReaderAdaptor::ArchiveReaderAdaptor(
     }
 }
 
+ArchiveReaderAdaptor::ArchiveReaderAdaptor(
+        std::shared_ptr<clp::ReaderInterface> single_file_archive_reader
+)
+        : m_single_file_archive{true},
+          m_timestamp_dictionary{std::make_shared<TimestampDictionaryReader>()},
+          m_reader{std::move(single_file_archive_reader)} {
+    if (nullptr == m_reader) {
+        throw OperationFailed(ErrorCodeBadParam, __FILENAME__, __LINE__);
+    }
+
+    if (auto const rc = m_reader->try_seek_from_begin(0); clp::ErrorCode::ErrorCode_Success != rc) {
+        throw OperationFailed(ErrorCodeFailure, __FILENAME__, __LINE__);
+    }
+}
+
 ErrorCode
 ArchiveReaderAdaptor::try_read_archive_file_info(ZstdDecompressor& decompressor, size_t size) {
     std::vector<char> buffer(size);
@@ -253,6 +269,10 @@ ErrorCode ArchiveReaderAdaptor::try_read_archive_metadata(ZstdDecompressor& deco
 }
 
 std::shared_ptr<clp::ReaderInterface> ArchiveReaderAdaptor::try_create_reader_at_header() {
+    if (nullptr != m_reader) {
+        return m_reader;
+    }
+
     if (InputSource::Filesystem == m_archive_path.source && false == m_single_file_archive) {
         try {
             return std::make_shared<clp::FileReader>(

components/core/src/clp_s/ArchiveReaderAdaptor.hpp

@@ -53,8 +53,20 @@ class ArchiveReaderAdaptor {
                 : TraceableException(error_code, filename, line_number) {}
     };
 
+    // Constructors
+    /**
+     * Creates an adaptor for an archive identified by path and source type.
+     * @param archive_path Path/URL for a directory archive or a single-file archive.
+     * @param network_auth Authentication options for network inputs.
+     */
     explicit ArchiveReaderAdaptor(Path const& archive_path, NetworkAuthOption const& network_auth);
 
+    /**
+     * Creates an adaptor around an already opened single-file archive reader.
+     * @param single_file_archive_reader An already-opened single-file archive stream.
+     */
+    explicit ArchiveReaderAdaptor(std::shared_ptr<clp::ReaderInterface> single_file_archive_reader);
+
     /**
      * Loads metadata for an archive including the header and metadata section. This method must be
      * invoked before checking out any section of an archive, or calling `get_timestamp_dictionary`.

docs/src/user-docs/guides-using-log-ingestor.md

@@ -51,12 +51,26 @@ will be routed through CLP's [API server](./guides-using-the-api-server.md) in a
 
 ### Fault tolerance
 
-:::{warning}
-**The current version of `log-ingestor` does not provide fault tolerance.**
+`log-ingestor` is designed to tolerate unexpected crashes or restarts without losing information
+about ingestion jobs or the files that have been submitted for compression. Note that this does not
+include fault tolerance of the components external to `log-ingestor`. Specifically, `log-ingestor`
+guarantees the following, even in the presence of crashes or restarts of `log-ingestor`:
 
-If `log-ingestor` crashes or is restarted, all in-progress ingestion jobs and their associated state
-will be lost, and must be restored manually. Robust fault tolerance for the ingestion pipeline is
-planned for a future release.
+* Any ingestion job successfully submitted to `log-ingestor` will run continuously.
+* Within an ingestion job, any files that have been found on S3 or received as messages from SQS
+  queue will eventually be submitted for compression.
+
+:::{note}
+`log-ingestor` **DOES NOT** guarantee the following after a crash or restart:
+
+* Any file submitted for compression (that can be compressed successfully) will eventually be
+  compressed successfully.
+  * This is because failures of the compression cluster are external to `log-ingestor`. Future
+    versions of CLP will address this limitation.
+* Any file submitted for compression will *only* be compressed once.
+  * This is because for [SQS listener](#sqs-listener) ingestion jobs, the processes for deleting
+    messages from the SQS queue and recording the files for ingestion are not synchronized. As a
+    result, a failure during this process may cause the same file to be ingested multiple times.
 :::
 
 ---