refactor(clp-package): Move archive metadata update from clp-s to compression task workers. (#819)

Author: Bill-hbrhbr

components/clp-py-utils/clp_py_utils/clp_config.py

@@ -36,6 +36,7 @@
 QUERY_TASKS_TABLE_NAME = "query_tasks"
 COMPRESSION_JOBS_TABLE_NAME = "compression_jobs"
 COMPRESSION_TASKS_TABLE_NAME = "compression_tasks"
+ARCHIVES_TABLE_SUFFIX = "archives"
 
 OS_RELEASE_FILE_PATH = pathlib.Path("etc") / "os-release"
 

components/core/src/clp_s/ArchiveWriter.cpp

@@ -6,6 +6,7 @@
 
 #include <json/single_include/nlohmann/json.hpp>
 
+#include "../clp/streaming_archive/Constants.hpp"
 #include "archive_constants.hpp"
 #include "Defs.hpp"
 #include "SchemaTree.hpp"
@@ -100,10 +101,6 @@ void ArchiveWriter::close() {
         header_and_metadata_writer.close();
     }
 
-    if (m_metadata_db) {
-        update_metadata_db();
-    }
-
     if (m_print_archive_stats) {
         print_archive_stats();
     }
@@ -430,29 +427,14 @@ std::pair<size_t, size_t> ArchiveWriter::store_tables() {
     return {table_metadata_compressed_size, table_compressed_size};
 }
 
-void ArchiveWriter::update_metadata_db() {
-    m_metadata_db->open();
-    clp::streaming_archive::ArchiveMetadata metadata(
-            cArchiveFormatDevelopmentVersionFlag,
-            "",
-            0ULL
-    );
-    metadata.increment_static_compressed_size(m_compressed_size);
-    metadata.increment_static_uncompressed_size(m_uncompressed_size);
-    metadata.expand_time_range(
-            m_timestamp_dict.get_begin_timestamp(),
-            m_timestamp_dict.get_end_timestamp()
-    );
-
-    m_metadata_db->add_archive(m_id, metadata);
-    m_metadata_db->close();
-}
-
-void ArchiveWriter::print_archive_stats() {
-    nlohmann::json json_msg;
-    json_msg["id"] = m_id;
-    json_msg["uncompressed_size"] = m_uncompressed_size;
-    json_msg["size"] = m_compressed_size;
+auto ArchiveWriter::print_archive_stats() const -> void {
+    namespace Archive = clp::streaming_archive::cMetadataDB::Archive;
+    nlohmann::json json_msg
+            = {{Archive::Id, m_id},
+               {Archive::BeginTimestamp, m_timestamp_dict.get_begin_timestamp()},
+               {Archive::EndTimestamp, m_timestamp_dict.get_end_timestamp()},
+               {Archive::UncompressedSize, m_uncompressed_size},
+               {Archive::Size, m_compressed_size}};
     std::cout << json_msg.dump(-1, ' ', true, nlohmann::json::error_handler_t::ignore) << std::endl;
 }
 }  // namespace clp_s

components/core/src/clp_s/ArchiveWriter.hpp

@@ -7,7 +7,6 @@
 #include <boost/uuid/uuid.hpp>
 #include <boost/uuid/uuid_io.hpp>
 
-#include "../clp/GlobalMySQLMetadataDB.hpp"
 #include "archive_constants.hpp"
 #include "DictionaryWriter.hpp"
 #include "Schema.hpp"
@@ -66,8 +65,7 @@ class ArchiveWriter {
     };
 
     // Constructor
-    explicit ArchiveWriter(std::shared_ptr<clp::GlobalMySQLMetadataDB> metadata_db)
-            : m_metadata_db(std::move(metadata_db)) {}
+    ArchiveWriter() = default;
 
     // Destructor
     ~ArchiveWriter() = default;
@@ -211,15 +209,10 @@ class ArchiveWriter {
      */
     void write_archive_header(FileWriter& archive_writer, size_t metadata_section_size);
 
-    /**
-     * Updates the metadata db with the archive's metadata (id, size, timestamp ranges, etc.)
-     */
-    void update_metadata_db();
-
     /**
      * Prints the archive's statistics (id, uncompressed size, compressed size, etc.)
      */
-    void print_archive_stats();
+    auto print_archive_stats() const -> void;
 
     static constexpr size_t cReadBlockSize = 4 * 1024;
 
@@ -238,7 +231,6 @@ class ArchiveWriter {
     std::shared_ptr<LogTypeDictionaryWriter> m_log_dict;
     std::shared_ptr<LogTypeDictionaryWriter> m_array_dict;  // log type dictionary for arrays
     TimestampDictionaryWriter m_timestamp_dict;
-    std::shared_ptr<clp::GlobalMySQLMetadataDB> m_metadata_db;
     int m_compression_level{};
     bool m_print_archive_stats{};
     bool m_single_file_archive{};

components/core/src/clp_s/CMakeLists.txt

@@ -18,8 +18,6 @@ set(
         ../clp/CurlStringList.hpp
         ../clp/cli_utils.cpp
         ../clp/cli_utils.hpp
-        ../clp/database_utils.cpp
-        ../clp/database_utils.hpp
         ../clp/Defs.h
         ../clp/ErrorCode.hpp
         ../clp/ffi/ir_stream/decoding_methods.cpp
@@ -42,23 +40,12 @@ set(
         ../clp/FileDescriptor.hpp
         ../clp/FileReader.cpp
         ../clp/FileReader.hpp
-        ../clp/GlobalMetadataDB.hpp
-        ../clp/GlobalMetadataDBConfig.cpp
-        ../clp/GlobalMetadataDBConfig.hpp
-        ../clp/GlobalMySQLMetadataDB.cpp
-        ../clp/GlobalMySQLMetadataDB.hpp
         ../clp/hash_utils.cpp
         ../clp/hash_utils.hpp
         ../clp/ir/EncodedTextAst.cpp
         ../clp/ir/EncodedTextAst.hpp
         ../clp/ir/parsing.cpp
         ../clp/ir/parsing.hpp
-        ../clp/MySQLDB.cpp
-        ../clp/MySQLDB.hpp
-        ../clp/MySQLParamBindings.cpp
-        ../clp/MySQLParamBindings.hpp
-        ../clp/MySQLPreparedStatement.cpp
-        ../clp/MySQLPreparedStatement.hpp
         ../clp/NetworkReader.cpp
         ../clp/NetworkReader.hpp
         ../clp/networking/socket_utils.cpp
@@ -70,6 +57,7 @@ set(
         ../clp/spdlog_with_specializations.hpp
         ../clp/streaming_archive/ArchiveMetadata.cpp
         ../clp/streaming_archive/ArchiveMetadata.hpp
+        ../clp/streaming_archive/Constants.hpp
         ../clp/streaming_compression/zstd/Decompressor.cpp
         ../clp/streaming_compression/zstd/Decompressor.hpp
         ../clp/Thread.cpp

components/core/src/clp_s/CommandLineArguments.cpp

@@ -205,7 +205,6 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) {
             // clang-format on
 
             po::options_description compression_options("Compression options");
-            std::string metadata_db_config_file_path;
             std::string input_path_list_file_path;
             constexpr std::string_view cJsonFileType{"json"};
             constexpr std::string_view cKeyValueIrFileType{"kv-ir"};
@@ -238,11 +237,6 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) {
                     po::value<std::string>(&m_timestamp_key)->value_name("TIMESTAMP_COLUMN_KEY")->
                         default_value(m_timestamp_key),
                     "Path (e.g. x.y) for the field containing the log event's timestamp."
-            )(
-                    "db-config-file",
-                    po::value<std::string>(&metadata_db_config_file_path)->value_name("FILE")->
-                    default_value(metadata_db_config_file_path),
-                    "Global metadata DB YAML config"
             )(
                     "files-from,f",
                     po::value<std::string>(&input_path_list_file_path)
@@ -353,29 +347,6 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) {
             }
 
             validate_network_auth(auth, m_network_auth);
-
-            // Parse and validate global metadata DB config
-            if (false == metadata_db_config_file_path.empty()) {
-                clp::GlobalMetadataDBConfig metadata_db_config;
-                try {
-                    metadata_db_config.parse_config_file(metadata_db_config_file_path);
-                } catch (std::exception& e) {
-                    SPDLOG_ERROR("Failed to validate metadata database config - {}.", e.what());
-                    return ParsingResult::Failure;
-                }
-
-                if (clp::GlobalMetadataDBConfig::MetadataDBType::MySQL
-                    != metadata_db_config.get_metadata_db_type())
-                {
-                    SPDLOG_ERROR(
-                            "Invalid metadata database type for {}; only supported type is MySQL.",
-                            m_program_name
-                    );
-                    return ParsingResult::Failure;
-                }
-
-                m_metadata_db_config = std::move(metadata_db_config);
-            }
         } else if ((char)Command::Extract == command_input) {
             po::options_description extraction_options;
             std::string archive_path;

components/core/src/clp_s/CommandLineArguments.hpp

@@ -9,7 +9,6 @@
 #include <boost/program_options/options_description.hpp>
 #include <boost/program_options/variables_map.hpp>
 
-#include "../clp/GlobalMetadataDBConfig.hpp"
 #include "../reducer/types.hpp"
 #include "Defs.hpp"
 #include "InputConfig.hpp"
@@ -94,10 +93,6 @@ class CommandLineArguments {
 
     bool get_ignore_case() const { return m_ignore_case; }
 
-    std::optional<clp::GlobalMetadataDBConfig> const& get_metadata_db_config() const {
-        return m_metadata_db_config;
-    }
-
     std::string const& get_reducer_host() const { return m_reducer_host; }
 
     int get_reducer_port() const { return m_reducer_port; }
@@ -200,9 +195,6 @@ class CommandLineArguments {
     bool m_disable_log_order{false};
     FileType m_file_type{FileType::Json};
 
-    // Metadata db variables
-    std::optional<clp::GlobalMetadataDBConfig> m_metadata_db_config;
-
     // MongoDB configuration variables
     std::string m_mongodb_uri;
     std::string m_mongodb_collection;

components/core/src/clp_s/JsonParser.cpp

@@ -127,7 +127,7 @@ JsonParser::JsonParser(JsonParserOption const& option)
     m_archive_options.authoritative_timestamp = m_timestamp_column;
     m_archive_options.authoritative_timestamp_namespace = m_timestamp_namespace;
 
-    m_archive_writer = std::make_unique<ArchiveWriter>(option.metadata_db);
+    m_archive_writer = std::make_unique<ArchiveWriter>();
     m_archive_writer->open(m_archive_options);
 }
 

components/core/src/clp_s/JsonParser.hpp

@@ -17,7 +17,6 @@
 #include "../clp/ffi/KeyValuePairLogEvent.hpp"
 #include "../clp/ffi/SchemaTree.hpp"
 #include "../clp/ffi/Value.hpp"
-#include "../clp/GlobalMySQLMetadataDB.hpp"
 #include "../clp/ReaderInterface.hpp"
 #include "ArchiveWriter.hpp"
 #include "CommandLineArguments.hpp"
@@ -51,7 +50,6 @@ struct JsonParserOption {
     bool structurize_arrays{};
     bool record_log_order{true};
     bool single_file_archive{false};
-    std::shared_ptr<clp::GlobalMySQLMetadataDB> metadata_db;
     NetworkAuthOption network_auth{};
 };
 

components/core/src/clp_s/clp-s.cpp

@@ -13,7 +13,6 @@
 #include <spdlog/spdlog.h>
 
 #include "../clp/CurlGlobalInstance.hpp"
-#include "../clp/GlobalMySQLMetadataDB.hpp"
 #include "../clp/streaming_archive/ArchiveMetadata.hpp"
 #include "../reducer/network_utils.hpp"
 #include "CommandLineArguments.hpp"
@@ -102,19 +101,6 @@ bool compress(CommandLineArguments const& command_line_arguments) {
     option.structurize_arrays = command_line_arguments.get_structurize_arrays();
     option.record_log_order = command_line_arguments.get_record_log_order();
 
-    auto const& db_config_container = command_line_arguments.get_metadata_db_config();
-    if (db_config_container.has_value()) {
-        auto const& db_config = db_config_container.value();
-        option.metadata_db = std::make_shared<clp::GlobalMySQLMetadataDB>(
-                db_config.get_metadata_db_host(),
-                db_config.get_metadata_db_port(),
-                db_config.get_metadata_db_username(),
-                db_config.get_metadata_db_password(),
-                db_config.get_metadata_db_name(),
-                db_config.get_metadata_table_prefix()
-        );
-    }
-
     clp_s::JsonParser parser(option);
     if (CommandLineArguments::FileType::KeyValueIr == option.input_file_type) {
         if (false == parser.parse_from_ir()) {

components/job-orchestration/job_orchestration/executor/compress/compression_task.py

@@ -10,6 +10,7 @@
 from celery.app.task import Task
 from celery.utils.log import get_task_logger
 from clp_py_utils.clp_config import (
+    ARCHIVES_TABLE_SUFFIX,
     COMPRESSION_JOBS_TABLE_NAME,
     COMPRESSION_TASKS_TABLE_NAME,
     Database,
@@ -82,6 +83,23 @@ def update_job_metadata_and_tags(db_cursor, job_id, table_prefix, tag_ids, archi
     )
 
 
+def update_archive_metadata(db_cursor, table_prefix, archive_stats):
+    archive_stats_defaults = {
+        "begin_timestamp": 0,
+        "end_timestamp": 0,
+        "creator_id": "",
+        "creation_ix": 0,
+    }
+    for k, v in archive_stats_defaults.items():
+        archive_stats.setdefault(k, v)
+    keys = ", ".join(archive_stats.keys())
+    value_placeholders = ", ".join(["%s"] * len(archive_stats))
+    query = (
+        f"INSERT INTO {table_prefix}{ARCHIVES_TABLE_SUFFIX} ({keys}) VALUES ({value_placeholders})"
+    )
+    db_cursor.execute(query, list(archive_stats.values()))
+
+
 def _generate_fs_logs_list(
     output_file_path: pathlib.Path,
     paths_to_compress: PathsToCompress,
@@ -161,15 +179,13 @@ def make_clp_s_command_and_env(
     clp_home: pathlib.Path,
     archive_output_dir: pathlib.Path,
     clp_config: ClpIoConfig,
-    db_config_file_path: pathlib.Path,
     use_single_file_archive: bool,
 ) -> Tuple[List[str], Optional[Dict[str, str]]]:
     """
     Generates the command and environment variables for a clp_s compression job.
     :param clp_home:
     :param archive_output_dir:
     :param clp_config:
-    :param db_config_file_path:
     :param use_single_file_archive:
     :return: Tuple of (compression_command, compression_env_vars)
     """
@@ -182,7 +198,6 @@ def make_clp_s_command_and_env(
         "--target-encoded-size",
         str(clp_config.output.target_segment_size + clp_config.output.target_dictionaries_size),
         "--compression-level", str(clp_config.output.compression_level),
-        "--db-config-file", str(db_config_file_path),
     ]
     # fmt: on
 
@@ -271,7 +286,6 @@ def run_clp(
             clp_home=clp_home,
             archive_output_dir=archive_output_dir,
             clp_config=clp_config,
-            db_config_file_path=db_config_file_path,
             use_single_file_archive=enable_s3_write,
         )
     else:
@@ -347,10 +361,13 @@ def run_clp(
                 with closing(sql_adapter.create_connection(True)) as db_conn, closing(
                     db_conn.cursor(dictionary=True)
                 ) as db_cursor:
+                    table_prefix = clp_metadata_db_connection_config["table_prefix"]
+                    if StorageEngine.CLP_S == clp_storage_engine:
+                        update_archive_metadata(db_cursor, table_prefix, last_archive_stats)
                     update_job_metadata_and_tags(
                         db_cursor,
                         job_id,
-                        clp_metadata_db_connection_config["table_prefix"],
+                        table_prefix,
                         tag_ids,
                         last_archive_stats,
                     )