diff --git a/components/clp-mcp-server/clp_mcp_server/clp_connector.py b/components/clp-mcp-server/clp_mcp_server/clp_connector.py index b3f158d419..8d95885121 100644 --- a/components/clp-mcp-server/clp_mcp_server/clp_connector.py +++ b/components/clp-mcp-server/clp_mcp_server/clp_connector.py @@ -161,11 +161,14 @@ async def read_results(self, query_id: str) -> list[dict]: results = [] async for doc in collection.find({}, limit=SEARCH_MAX_NUM_RESULTS): + result_id = doc["_id"] + doc["archive_id"] = result_id["archive_id"] + doc["log_event_idx"] = result_id["log_event_idx"] doc["link"] = ( f"{self._webui_addr}/streamFile?type=json" f"&streamId={doc['archive_id']}" f"&dataset={CLP_DEFAULT_DATASET_NAME}" - f"&logEventIdx={doc['log_event_ix']}" + f"&logEventIdx={doc['log_event_idx']}" ) doc["_id"] = None results.append(doc) diff --git a/components/clp-mcp-server/tests/test_clp_connector.py b/components/clp-mcp-server/tests/test_clp_connector.py index dd5ff60115..59b31c6aaa 100644 --- a/components/clp-mcp-server/tests/test_clp_connector.py +++ b/components/clp-mcp-server/tests/test_clp_connector.py @@ -111,9 +111,9 @@ async def test_read_results_returns_docs(mock_clp_config: Any) -> None: """Tests reading results returns expected documents.""" connector = ClpConnector(mock_clp_config) mock_docs = [ - {"_id": "1", "archive_id": "archA", "log_event_ix": 1}, - {"_id": "2", "archive_id": "archB", "log_event_ix": 2}, - {"_id": "3", "archive_id": "archC", "log_event_ix": 3}, + {"_id": {"archive_id": "archA", "log_event_idx": 1}}, + {"_id": {"archive_id": "archB", "log_event_idx": 2}}, + {"_id": {"archive_id": "archC", "log_event_idx": 3}}, ] mock_collection = AsyncMock() mock_collection.find = MagicMock(return_value=_aiter(mock_docs)) @@ -129,8 +129,8 @@ async def test_read_results_adds_link_field(mock_clp_config: Any) -> None: """Ensures read_results adds a 'link' field.""" connector = ClpConnector(mock_clp_config) mock_docs = [ - {"_id": "1", "archive_id": "archA", "log_event_ix": 10}, - {"_id": "2", "archive_id": "archB", "log_event_ix": 20}, + {"_id": {"archive_id": "archA", "log_event_idx": 10}}, + {"_id": {"archive_id": "archB", "log_event_idx": 20}}, ] mock_collection = AsyncMock() mock_collection.find = MagicMock(return_value=_aiter(mock_docs)) @@ -143,7 +143,7 @@ async def test_read_results_adds_link_field(mock_clp_config: Any) -> None: expected_link = ( f"http://{mock_clp_config.webui.host}:{mock_clp_config.webui.port}" f"/streamFile?type=json&streamId={original['archive_id']}" - f"&dataset=default&logEventIdx={original['log_event_ix']}" + f"&dataset=default&logEventIdx={original['log_event_idx']}" ) assert result["link"] == expected_link diff --git a/components/core/src/clp/clo/CMakeLists.txt b/components/core/src/clp/clo/CMakeLists.txt index 588267dfbb..f567e2044a 100644 --- a/components/core/src/clp/clo/CMakeLists.txt +++ b/components/core/src/clp/clo/CMakeLists.txt @@ -166,8 +166,14 @@ set( ../../reducer/types.hpp ) +set( + CLP_S_MONGODB_UTILS_SOURCES + ../../clp_s/MongoDBUtils.cpp + ../../clp_s/MongoDBUtils.hpp +) + if(CLP_BUILD_EXECUTABLES) - add_executable(clo ${CLO_SOURCES} ${REDUCER_SOURCES}) + add_executable(clo ${CLO_SOURCES} ${CLP_S_MONGODB_UTILS_SOURCES} ${REDUCER_SOURCES}) target_compile_features(clo PRIVATE cxx_std_20) target_include_directories(clo PRIVATE diff --git a/components/core/src/clp/clo/OutputHandler.cpp b/components/core/src/clp/clo/OutputHandler.cpp index 1ad61221c2..c56d57b92a 100644 --- a/components/core/src/clp/clo/OutputHandler.cpp +++ b/components/core/src/clp/clo/OutputHandler.cpp @@ -4,9 +4,12 @@ #include #include +#include #include #include +#include + #include "../../reducer/CountOperator.hpp" #include "../../reducer/network_utils.hpp" #include "../networking/socket_utils.hpp" @@ -51,6 +54,7 @@ ResultsCacheOutputHandler::ResultsCacheOutputHandler( ) : m_batch_size(batch_size), m_max_num_results(max_num_results) { + m_insert_options.ordered(false); try { auto mongo_uri = mongocxx::uri(uri); m_client = mongocxx::client(mongo_uri); @@ -106,17 +110,24 @@ ErrorCode ResultsCacheOutputHandler::flush() { std::move( bsoncxx::builder::basic::make_document( bsoncxx::builder::basic::kvp( - cResultsCacheKeys::SearchOutput::OrigFileId, - std::move(result.orig_file_id) + cResultsCacheKeys::SearchOutput::Id, + bsoncxx::builder::basic::make_document( + bsoncxx::builder::basic::kvp( + cResultsCacheKeys::SearchOutput:: + OrigFileId, + result.orig_file_id + ), + bsoncxx::builder::basic::kvp( + cResultsCacheKeys::SearchOutput:: + LogEventIdx, + result.log_event_ix + ) + ) ), bsoncxx::builder::basic::kvp( cResultsCacheKeys::SearchOutput::OrigFilePath, std::move(result.orig_file_path) ), - bsoncxx::builder::basic::kvp( - cResultsCacheKeys::SearchOutput::LogEventIx, - result.log_event_ix - ), bsoncxx::builder::basic::kvp( cResultsCacheKeys::SearchOutput::Timestamp, result.timestamp @@ -128,29 +139,42 @@ ErrorCode ResultsCacheOutputHandler::flush() { ) ) ); - count++; - - if (count == m_batch_size) { - m_collection.insert_many(m_results); - m_results.clear(); - count = 0; - } } catch (mongocxx::exception const& e) { + SPDLOG_ERROR("Failed to build search result - {}", e.what()); return ErrorCode::ErrorCode_Failure_DB_Bulk_Write; } - } - try { - if (false == m_results.empty()) { - m_collection.insert_many(m_results); - m_results.clear(); + count++; + if (count == m_batch_size) { + if (false == insert_results()) { + return ErrorCode::ErrorCode_Failure_DB_Bulk_Write; + } + count = 0; } - } catch (mongocxx::exception const& e) { + } + + if (false == m_results.empty() && false == insert_results()) { return ErrorCode::ErrorCode_Failure_DB_Bulk_Write; } return ErrorCode::ErrorCode_Success; } +auto ResultsCacheOutputHandler::insert_results() -> bool { + try { + m_collection.insert_many(m_results, m_insert_options); + } catch (mongocxx::bulk_write_exception const& exception) { + if (false == clp_s::contains_only_duplicate_key_write_errors(exception, m_results.size())) { + SPDLOG_ERROR("Failed to insert search results - {}", exception.what()); + return false; + } + } catch (mongocxx::exception const& exception) { + SPDLOG_ERROR("Failed to insert search results - {}", exception.what()); + return false; + } + m_results.clear(); + return true; +} + CountOutputHandler::CountOutputHandler(int reducer_socket_fd) : m_reducer_socket_fd{reducer_socket_fd}, m_pipeline{reducer::PipelineInputMode::InterStage} { diff --git a/components/core/src/clp/clo/OutputHandler.hpp b/components/core/src/clp/clo/OutputHandler.hpp index b4824b5202..7611efd086 100644 --- a/components/core/src/clp/clo/OutputHandler.hpp +++ b/components/core/src/clp/clo/OutputHandler.hpp @@ -10,6 +10,7 @@ #include #include #include +#include #include #include "../../reducer/Pipeline.hpp" @@ -208,8 +209,16 @@ class ResultsCacheOutputHandler : public OutputHandler { return m_latest_results.size() >= m_max_num_results; } + /** + * Inserts the pending results as an unordered batch. Duplicate-key errors are treated as + * success so that retries converge on the complete result set. + * @return Whether insertion succeeded or produced only duplicate-key errors. + */ + [[nodiscard]] auto insert_results() -> bool; + mongocxx::client m_client; mongocxx::collection m_collection; + mongocxx::options::insert m_insert_options; std::vector m_results; uint64_t m_batch_size; uint64_t m_max_num_results; diff --git a/components/core/src/clp/clo/constants.hpp b/components/core/src/clp/clo/constants.hpp index 945bde83e0..295179e220 100644 --- a/components/core/src/clp/clo/constants.hpp +++ b/components/core/src/clp/clo/constants.hpp @@ -12,9 +12,10 @@ constexpr char IsLastChunk[]{"is_last_chunk"}; } // namespace IrOutput namespace SearchOutput { +constexpr char Id[]{"_id"}; constexpr char OrigFileId[]{"orig_file_id"}; constexpr char OrigFilePath[]{"orig_file_path"}; -constexpr char LogEventIx[]{"log_event_ix"}; +constexpr char LogEventIdx[]{"log_event_idx"}; constexpr char Timestamp[]{"timestamp"}; constexpr char Message[]{"message"}; } // namespace SearchOutput diff --git a/components/core/src/clp_s/CMakeLists.txt b/components/core/src/clp_s/CMakeLists.txt index b845622bbd..d0a625a6e9 100644 --- a/components/core/src/clp_s/CMakeLists.txt +++ b/components/core/src/clp_s/CMakeLists.txt @@ -492,6 +492,8 @@ set( ErrorCode.hpp kv_ir_search.cpp kv_ir_search.hpp + MongoDBUtils.cpp + MongoDBUtils.hpp OutputHandlerImpl.cpp OutputHandlerImpl.hpp ResultsCacheUtils.cpp diff --git a/components/core/src/clp_s/MongoDBUtils.cpp b/components/core/src/clp_s/MongoDBUtils.cpp new file mode 100644 index 0000000000..990771d62f --- /dev/null +++ b/components/core/src/clp_s/MongoDBUtils.cpp @@ -0,0 +1,128 @@ +#include "MongoDBUtils.hpp" + +#include + +#include +#include + +namespace clp_s { +namespace { +constexpr int32_t cDuplicateKeyErrorCode{11'000}; + +/** + * Checks whether an aggregated bulk-write reply contains any command errors. + * @param reply The raw MongoDB bulk-write reply. + * @return Whether the reply contains a command error. + */ +[[nodiscard]] auto has_command_errors(bsoncxx::document::view const& reply) -> bool; + +/** + * Checks whether a bulk-write reply contains any write-concern errors. + * @param reply The raw MongoDB bulk-write reply. + * @return Whether the reply contains a write-concern error. + */ +[[nodiscard]] auto has_write_concern_errors(bsoncxx::document::view const& reply) -> bool; + +/** + * Checks whether an entry from a bulk-write reply's `writeErrors` array is a duplicate-key error. + * @param write_error The write-error entry to inspect. + * @return Whether the entry has MongoDB's duplicate-key error code. + */ +[[nodiscard]] auto is_duplicate_key_write_error(bsoncxx::array::element const& write_error) -> bool; + +[[nodiscard]] auto has_command_errors(bsoncxx::document::view const& reply) -> bool { + auto const errors_element = reply["errorReplies"]; + if (false == static_cast(errors_element)) { + return false; + } + if (bsoncxx::type::k_array != errors_element.type()) { + return true; + } + auto const errors = errors_element.get_array().value; + return false == errors.empty(); +} + +[[nodiscard]] auto has_write_concern_errors(bsoncxx::document::view const& reply) -> bool { + if (static_cast(reply["writeConcernError"])) { + return true; + } + + auto const errors_element = reply["writeConcernErrors"]; + if (false == static_cast(errors_element)) { + return false; + } + if (bsoncxx::type::k_array != errors_element.type()) { + return true; + } + auto const errors = errors_element.get_array().value; + return false == errors.empty(); +} + +[[nodiscard]] auto is_duplicate_key_write_error(bsoncxx::array::element const& write_error) + -> bool { + if (bsoncxx::type::k_document != write_error.type()) { + return false; + } + + auto const code = write_error.get_document().value["code"]; + if (false == static_cast(code)) { + return false; + } + if (bsoncxx::type::k_int32 == code.type()) { + return cDuplicateKeyErrorCode == code.get_int32().value; + } + if (bsoncxx::type::k_int64 == code.type()) { + return cDuplicateKeyErrorCode == code.get_int64().value; + } + return false; +} +} // namespace + +auto contains_only_duplicate_key_write_errors( + mongocxx::bulk_write_exception const& exception, + size_t num_documents +) -> bool { + auto const& raw_server_error = exception.raw_server_error(); + if (false == raw_server_error.has_value()) { + return false; + } + + auto const reply = raw_server_error->view(); + if (has_command_errors(reply) || has_write_concern_errors(reply)) { + return false; + } + + auto const write_errors_element = reply["writeErrors"]; + if (false == static_cast(write_errors_element) + || bsoncxx::type::k_array != write_errors_element.type()) + { + return false; + } + + auto const write_errors = write_errors_element.get_array().value; + if (write_errors.empty()) { + return false; + } + + size_t num_write_errors{0}; + for (auto const& write_error : write_errors) { + if (false == is_duplicate_key_write_error(write_error)) { + return false; + } + ++num_write_errors; + } + + auto const num_inserted_element = reply["nInserted"]; + if (false == static_cast(num_inserted_element) + || bsoncxx::type::k_int32 != num_inserted_element.type()) + { + return false; + } + auto const num_inserted = num_inserted_element.get_int32().value; + if (num_inserted < 0) { + return false; + } + + return static_cast(num_inserted) + num_write_errors == num_documents; +} +} // namespace clp_s diff --git a/components/core/src/clp_s/MongoDBUtils.hpp b/components/core/src/clp_s/MongoDBUtils.hpp new file mode 100644 index 0000000000..993d31ca81 --- /dev/null +++ b/components/core/src/clp_s/MongoDBUtils.hpp @@ -0,0 +1,26 @@ +#ifndef CLP_S_MONGODBUTILS_HPP +#define CLP_S_MONGODBUTILS_HPP + +#include + +#include + +namespace clp_s { +/** + * Returns whether the bulk write failed only because some documents already exist. + * + * Command and write-concern errors are rejected since they mean MongoDB did not confirm the outcome + * of the entire batch. The number of inserted documents and write errors must account for every + * submitted document, and every write error must be a duplicate-key error. + * @param exception The exception containing the raw MongoDB bulk-write reply. + * @param num_documents The number of documents submitted in the bulk write. + * @return Whether the reply accounts for every document using successful inserts and duplicate-key + * errors only. + */ +[[nodiscard]] auto contains_only_duplicate_key_write_errors( + mongocxx::bulk_write_exception const& exception, + size_t num_documents +) -> bool; +} // namespace clp_s + +#endif // CLP_S_MONGODBUTILS_HPP diff --git a/components/core/src/clp_s/OutputHandlerImpl.cpp b/components/core/src/clp_s/OutputHandlerImpl.cpp index e2ebbe48f2..5730ce84b2 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.cpp +++ b/components/core/src/clp_s/OutputHandlerImpl.cpp @@ -9,11 +9,13 @@ #include #include #include +#include #include #include #include #include +#include #include #include "../clp/networking/socket_utils.hpp" @@ -83,6 +85,7 @@ ResultsCacheOutputHandler::ResultsCacheOutputHandler( m_max_num_results{max_num_results}, m_dataset{dataset} { m_collection = connect_to_results_cache(uri, collection, m_client); + m_insert_options.ordered(false); m_results.reserve(m_batch_size); } @@ -96,6 +99,21 @@ ErrorCode ResultsCacheOutputHandler::finish() { m_results.emplace_back( std::move( bsoncxx::builder::basic::make_document( + bsoncxx::builder::basic::kvp( + constants::results_cache::search::cId, + bsoncxx::builder::basic::make_document( + bsoncxx::builder::basic::kvp( + constants::results_cache::search:: + cArchiveId, + std::move(result.archive_id) + ), + bsoncxx::builder::basic::kvp( + constants::results_cache::search:: + cLogEventIdx, + result.log_event_idx + ) + ) + ), bsoncxx::builder::basic::kvp( constants::results_cache::search::cOrigFilePath, std::move(result.original_path) @@ -108,14 +126,6 @@ ErrorCode ResultsCacheOutputHandler::finish() { constants::results_cache::search::cTimestamp, result.timestamp ), - bsoncxx::builder::basic::kvp( - constants::results_cache::search::cArchiveId, - std::move(result.archive_id) - ), - bsoncxx::builder::basic::kvp( - constants::results_cache::search::cLogEventIx, - result.log_event_idx - ), bsoncxx::builder::basic::kvp( std::string{constants::results_cache::search::cDataset}, std::move(result.dataset) @@ -123,24 +133,21 @@ ErrorCode ResultsCacheOutputHandler::finish() { ) ) ); - count++; - - if (count == m_batch_size) { - m_collection.insert_many(m_results); - m_results.clear(); - count = 0; - } } catch (mongocxx::exception const& e) { + SPDLOG_ERROR("Failed to build search result - {}", e.what()); return ErrorCode::ErrorCodeFailureDbBulkWrite; } - } - try { - if (false == m_results.empty()) { - m_collection.insert_many(m_results); - m_results.clear(); + count++; + if (count == m_batch_size) { + if (false == insert_results()) { + return ErrorCode::ErrorCodeFailureDbBulkWrite; + } + count = 0; } - } catch (mongocxx::exception const& e) { + } + + if (false == m_results.empty() && false == insert_results()) { return ErrorCode::ErrorCodeFailureDbBulkWrite; } return ErrorCode::ErrorCodeSuccess; @@ -178,6 +185,22 @@ void ResultsCacheOutputHandler::write( } } +auto ResultsCacheOutputHandler::insert_results() -> bool { + try { + m_collection.insert_many(m_results, m_insert_options); + } catch (mongocxx::bulk_write_exception const& exception) { + if (false == contains_only_duplicate_key_write_errors(exception, m_results.size())) { + SPDLOG_ERROR("Failed to insert search results - {}", exception.what()); + return false; + } + } catch (mongocxx::exception const& exception) { + SPDLOG_ERROR("Failed to insert search results - {}", exception.what()); + return false; + } + m_results.clear(); + return true; +} + CountReducerOutputHandler::CountReducerOutputHandler(int reducer_socket_fd) : search::OutputHandler(false, false), m_reducer_socket_fd(reducer_socket_fd), diff --git a/components/core/src/clp_s/OutputHandlerImpl.hpp b/components/core/src/clp_s/OutputHandlerImpl.hpp index 5cacff8966..8afb3a1587 100644 --- a/components/core/src/clp_s/OutputHandlerImpl.hpp +++ b/components/core/src/clp_s/OutputHandlerImpl.hpp @@ -18,6 +18,7 @@ #include #include +#include #include #include @@ -200,8 +201,16 @@ class ResultsCacheOutputHandler : public ::clp_s::search::OutputHandler { void write(std::string_view message) override { write(message, 0, {}, 0); } private: + /** + * Inserts the pending results as an unordered batch. Duplicate-key errors are treated as + * success so that retries converge on the complete result set. + * @return Whether insertion succeeded or produced only duplicate-key errors. + */ + [[nodiscard]] auto insert_results() -> bool; + mongocxx::client m_client; mongocxx::collection m_collection; + mongocxx::options::insert m_insert_options; std::vector m_results; uint64_t m_batch_size; uint64_t m_max_num_results; diff --git a/components/core/src/clp_s/archive_constants.hpp b/components/core/src/clp_s/archive_constants.hpp index 0dbf3d4c68..9d8b134842 100644 --- a/components/core/src/clp_s/archive_constants.hpp +++ b/components/core/src/clp_s/archive_constants.hpp @@ -53,8 +53,9 @@ constexpr char cIsLastChunk[]{"is_last_chunk"}; } // namespace results_cache::decompression namespace results_cache::search { +constexpr char cId[]{"_id"}; constexpr char cOrigFilePath[]{"orig_file_path"}; -constexpr char cLogEventIx[]{"log_event_ix"}; +constexpr char cLogEventIdx[]{"log_event_idx"}; constexpr char cTimestamp[]{"timestamp"}; constexpr char cMessage[]{"message"}; constexpr char cArchiveId[]{"archive_id"}; diff --git a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py index 72c24e76f2..9e4689f6f6 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/search_result_garbage_collector.py @@ -1,33 +1,90 @@ +"""Garbage-collect cached search results.""" + import asyncio -from typing import Final +from contextlib import closing +from typing import Any, cast, Final import pymongo import pymongo.database -from bson import ObjectId -from clp_py_utils.clp_config import ClpConfig, ResultsCache +from clp_py_utils.clp_config import ( + ClpConfig, + Database, + QUERY_JOBS_TABLE_NAME, + ResultsCache, +) from clp_py_utils.clp_logging import configure_logging, get_logger +from clp_py_utils.sql_adapter import SqlAdapter from job_orchestration.garbage_collector.constants import ( MIN_TO_SECONDS, SEARCH_RESULT_GARBAGE_COLLECTOR_NAME, ) -from job_orchestration.garbage_collector.utils import get_expiry_epoch_secs +from job_orchestration.scheduler.constants import QueryJobStatus # Constants MONGODB_ID_KEY: Final[str] = "_id" +MAX_NUM_JOB_IDS_PER_QUERY: Final[int] = 1000 logger = get_logger(SEARCH_RESULT_GARBAGE_COLLECTOR_NAME) -def _get_latest_doc_timestamp(collection: pymongo.collection.Collection) -> int: - latest_doc = collection.find_one(sort=[(MONGODB_ID_KEY, pymongo.DESCENDING)]) - if latest_doc is None: - return 0 - - object_id = latest_doc[MONGODB_ID_KEY] - if isinstance(object_id, ObjectId): - return int(object_id.generation_time.timestamp()) - raise ValueError(f"{object_id} is not an ObjectID") +def _get_expired_job_ids( + database_config: Database, job_ids: list[int], retention_period_minutes: int +) -> list[int]: + """ + Filters query job IDs by whether their retention periods have ended. + + MariaDB computes each query's completion time as `start_time + duration`. A query job ID is + included when the time since completion is greater than `retention_period_minutes`. For a + terminated query without a completion time, `creation_time` is used instead. + + :param database_config: Configuration for the orchestration database. + :param job_ids: Query job IDs to filter. + :param retention_period_minutes: Length of the retention period following query completion, in + minutes. + :return: Query job IDs completed more than `retention_period_minutes` ago. + """ + if len(job_ids) == 0: + return [] + + expired_job_ids: list[int] = [] + sql_adapter = SqlAdapter(database_config) + with ( + closing(sql_adapter.create_connection(True)) as db_conn, + closing(db_conn.cursor()) as db_cursor, + ): + for begin_idx in range(0, len(job_ids), MAX_NUM_JOB_IDS_PER_QUERY): + job_ids_batch = job_ids[begin_idx : begin_idx + MAX_NUM_JOB_IDS_PER_QUERY] + job_id_placeholders = ",".join(["%s"] * len(job_ids_batch)) + query = f""" + SELECT id + FROM `{QUERY_JOBS_TABLE_NAME}` + WHERE id IN ({job_id_placeholders}) + AND ( + TIMESTAMPADD( + MICROSECOND, + CAST(duration * 1000000 AS SIGNED), + start_time + ) < TIMESTAMPADD(MINUTE, %s, CURRENT_TIMESTAMP(3)) + OR ( + (start_time IS NULL OR duration IS NULL) + AND status IN ( + {QueryJobStatus.SUCCEEDED}, + {QueryJobStatus.FAILED}, + {QueryJobStatus.CANCELLED}, + {QueryJobStatus.KILLED} + ) + AND creation_time < TIMESTAMPADD(MINUTE, %s, CURRENT_TIMESTAMP(3)) + ) + ) + """ + db_cursor.execute( + query, + [*job_ids_batch, -retention_period_minutes, -retention_period_minutes], + ) + rows = cast("list[tuple[int]]", db_cursor.fetchall()) + expired_job_ids.extend(row[0] for row in rows) + return expired_job_ids def _delete_result_metadata( @@ -38,27 +95,37 @@ def _delete_result_metadata( def _collect_and_sweep_expired_search_results( - result_cache_config: ResultsCache, results_metadata_collection_name: str -): - expiry_epoch = get_expiry_epoch_secs(result_cache_config.retention_period) + result_cache_config: ResultsCache, + database_config: Database, + results_metadata_collection_name: str, +) -> None: + """ + Removes search results whose query completion time is older than the retention cutoff. - logger.debug(f"Searching for search jobs finished before {expiry_epoch}.") - deleted_job_ids: list[int] = [] - with pymongo.MongoClient(result_cache_config.get_uri()) as results_cache_client: - results_cache_db = results_cache_client.get_default_database() - collection_names = results_cache_db.list_collection_names() - for job_id in collection_names: - if not job_id.isdigit(): - continue + Numeric MongoDB collection names are interpreted as query job IDs. Collections selected by + `_get_expired_job_ids` are dropped along with their result metadata documents. - job_results_collection = results_cache_db.get_collection(job_id) - collection_timestamp = _get_latest_doc_timestamp(job_results_collection) - if collection_timestamp >= expiry_epoch: - continue + :param result_cache_config: MongoDB result-cache and retention configuration. + :param database_config: Configuration for the orchestration database. + :param results_metadata_collection_name: Name of the result metadata collection. + """ + retention_period = result_cache_config.retention_period + if retention_period is None: + return - _delete_result_metadata(results_cache_db, results_metadata_collection_name, job_id) - job_results_collection.drop() - deleted_job_ids.append(int(job_id)) + deleted_job_ids: list[int] = [] + results_cache_client: pymongo.MongoClient[dict[str, Any]] = pymongo.MongoClient( + result_cache_config.get_uri() + ) + with results_cache_client: + results_cache_db = results_cache_client.get_default_database() + job_ids = [int(name) for name in results_cache_db.list_collection_names() if name.isdigit()] + expired_job_ids = _get_expired_job_ids(database_config, job_ids, retention_period) + for job_id in expired_job_ids: + job_id_str = str(job_id) + _delete_result_metadata(results_cache_db, results_metadata_collection_name, job_id_str) + results_cache_db.get_collection(job_id_str).drop() + deleted_job_ids.append(job_id) if len(deleted_job_ids) != 0: logger.debug(f"Deleted search results of job(s): {deleted_job_ids}.") @@ -67,6 +134,7 @@ def _collect_and_sweep_expired_search_results( async def search_result_garbage_collector(clp_config: ClpConfig) -> None: + """Run search-result collection and sweeping at the configured interval.""" configure_logging(logger, SEARCH_RESULT_GARBAGE_COLLECTOR_NAME) sweep_interval_secs = clp_config.garbage_collector.sweep_interval.search_result * MIN_TO_SECONDS @@ -75,7 +143,9 @@ async def search_result_garbage_collector(clp_config: ClpConfig) -> None: try: while True: _collect_and_sweep_expired_search_results( - clp_config.results_cache, clp_config.webui.results_metadata_collection_name + clp_config.results_cache, + clp_config.database, + clp_config.webui.results_metadata_collection_name, ) await asyncio.sleep(sweep_interval_secs) except Exception: diff --git a/components/job-orchestration/job_orchestration/garbage_collector/utils.py b/components/job-orchestration/job_orchestration/garbage_collector/utils.py index a53a0715a2..45aa31188d 100644 --- a/components/job-orchestration/job_orchestration/garbage_collector/utils.py +++ b/components/job-orchestration/job_orchestration/garbage_collector/utils.py @@ -1,7 +1,6 @@ import os import pathlib import shutil -import time from datetime import datetime, timezone from bson import ObjectId @@ -13,8 +12,6 @@ ) from clp_py_utils.s3_utils import s3_delete_objects -from job_orchestration.garbage_collector.constants import MIN_TO_SECONDS - def validate_storage_type(output_config: ArchiveOutput, storage_engine: str) -> None: storage_type = output_config.storage.type @@ -27,18 +24,6 @@ def validate_storage_type(output_config: ArchiveOutput, storage_engine: str) -> raise ValueError(f"Unsupported Storage type: {storage_type}") -def get_expiry_epoch_secs(retention_minutes: int) -> int: - """ - Returns a cutoff `expiry_epoch` based on the current timestamp and `retention_minutes`. Any - candidate with a timestamp (`ts`) less than `expiry_epoch` is considered expired. - The `expiry_epoch` is calculated as `expiry_epoch = cur_time - retention_secs`. - - :param retention_minutes: Retention period in minutes. - :return: The UTC epoch representing the expiry cutoff time. - """ - return int(time.time() - retention_minutes * MIN_TO_SECONDS) - - def get_oid_with_expiry_time(expiry_epoch_secs: int) -> ObjectId: return ObjectId.from_datetime(datetime.fromtimestamp(expiry_epoch_secs, tz=timezone.utc)) diff --git a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/typings.tsx b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/typings.tsx index 41852da995..f0b1177856 100644 --- a/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/typings.tsx +++ b/components/webui/packages/client/src/pages/SearchPage/SearchResults/SearchResultsTable/Native/SearchResultsVirtualTable/typings.tsx @@ -25,7 +25,7 @@ interface SearchResult { archive_id: string; dataset: string; filePath: string; - log_event_ix: number; + log_event_idx: number; message: string; orig_file_id: string; orig_file_path: string; @@ -90,7 +90,7 @@ const searchResultsTableColumns: NonNullable["columns"] render: (_, record) => ( ; + +type RawSearchResult = + (SearchResultWithoutId & { + _id: ClpSearchResultId; + }) | + (SearchResultWithoutId & { + _id: ClpSSearchResultId; + }); + + /** * Custom hook to stream search results for the current searchJobId from the API server's SSE * endpoint. When the stream ends, the search UI state is updated to `DONE` (or `FAILED` if the @@ -33,15 +57,24 @@ const useSearchResults = () => { } }, parse: (data) => { - // MongoDB ObjectIds are serialized as `{"$oid": "..."}` in the SSE stream. - const doc = JSON.parse(data) as Omit & - {_id: string | {$oid: string}}; + const doc = JSON.parse(data) as RawSearchResult; + + if ("orig_file_id" in doc._id) { + return { + ...doc, + _id: JSON.stringify(doc._id), + archive_id: "", + log_event_idx: doc._id.log_event_idx, + orig_file_id: doc._id.orig_file_id, + }; + } return { ...doc, - _id: "object" === typeof doc._id ? - doc._id.$oid : - doc._id, + _id: JSON.stringify(doc._id), + archive_id: doc._id.archive_id, + log_event_idx: doc._id.log_event_idx, + orig_file_id: "", }; }, rawDocs: true,