feat(clp-package): Expose dataset selection in package CLI scripts and web UI. (Deprecated)

components/clp-package-utils/clp_package_utils/scripts/archive_manager.py

@@ -5,7 +5,10 @@
 import typing
 from pathlib import Path
 
-from clp_py_utils.clp_config import StorageType
+from clp_py_utils.clp_config import (
+    CLP_DEFAULT_DATASET_NAME,
+    StorageType,
+)
 
 from clp_package_utils.general import (
     CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH,
@@ -61,6 +64,11 @@ def main(argv: typing.List[str]) -> int:
         default=str(default_config_file_path),
         help="CLP package configuration file.",
     )
+    args_parser.add_argument(
+        "--dataset",
+        default=CLP_DEFAULT_DATASET_NAME,
+        help="The name of the log category.",
+    )
 
     # Top-level commands
     subparsers: argparse._SubParsersAction[argparse.ArgumentParser] = args_parser.add_subparsers(
@@ -196,6 +204,7 @@ def main(argv: typing.List[str]) -> int:
         "python3",
         "-m", "clp_package_utils.scripts.native.archive_manager",
         "--config", str(generated_config_path_on_container),
+        "--dataset", str(parsed_args.dataset),
         str(subcommand),
     ]
     # fmt : on

components/clp-package-utils/clp_package_utils/scripts/compress.py

@@ -6,7 +6,10 @@
 import uuid
 from typing import List
 
-from clp_py_utils.clp_config import StorageEngine
+from clp_py_utils.clp_config import (
+    CLP_DEFAULT_DATASET_NAME,
+    StorageEngine,
+)
 from job_orchestration.scheduler.job_config import InputType
 
 from clp_package_utils.general import (
@@ -74,6 +77,8 @@ def _generate_compress_cmd(
         "--config", str(config_path),
     ]
     # fmt: on
+    compress_cmd.append("--dataset")
+    compress_cmd.append(str(parsed_args.dataset))
     if parsed_args.timestamp_key is not None:
         compress_cmd.append("--timestamp-key")
         compress_cmd.append(parsed_args.timestamp_key)
@@ -82,7 +87,6 @@ def _generate_compress_cmd(
         compress_cmd.append(parsed_args.tags)
     if parsed_args.no_progress_reporting is True:
         compress_cmd.append("--no-progress-reporting")
-
     compress_cmd.append("--logs-list")
     compress_cmd.append(str(logs_list_path))
 
@@ -131,6 +135,11 @@ def main(argv):
         default=str(default_config_file_path),
         help="CLP package configuration file.",
     )
+    args_parser.add_argument(
+        "--dataset",
+        default=CLP_DEFAULT_DATASET_NAME,
+        help="The name of the log category.",
+    )
     args_parser.add_argument(
         "--timestamp-key",
         help="The path (e.g. x.y) for the field containing the log event's timestamp.",

components/clp-package-utils/clp_package_utils/scripts/decompress.py

@@ -5,7 +5,12 @@
 import sys
 from typing import Optional
 
-from clp_py_utils.clp_config import CLPConfig, StorageEngine, StorageType
+from clp_py_utils.clp_config import (
+    CLP_DEFAULT_DATASET_NAME,
+    CLPConfig,
+    StorageEngine,
+    StorageType,
+)
 
 from clp_package_utils.general import (
     CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH,
@@ -207,6 +212,8 @@ def handle_extract_stream_cmd(
             extract_cmd.append(str(parsed_args.target_uncompressed_size))
     elif EXTRACT_JSON_CMD == job_command:
         extract_cmd.append(str(parsed_args.archive_id))
+        extract_cmd.append("--dataset")
+        extract_cmd.append(str(parsed_args.dataset))
         if parsed_args.target_chunk_size:
             extract_cmd.append("--target-chunk-size")
             extract_cmd.append(str(parsed_args.target_chunk_size))
@@ -267,6 +274,12 @@ def main(argv):
     # JSON extraction command parser
     json_extraction_parser = command_args_parser.add_parser(EXTRACT_JSON_CMD)
     json_extraction_parser.add_argument("archive_id", type=str, help="Archive ID")
+    json_extraction_parser.add_argument(
+        "--dataset",
+        type=str,
+        default=CLP_DEFAULT_DATASET_NAME,
+        help="The name of the log category.",
+    )
     json_extraction_parser.add_argument(
         "--target-chunk-size",
         type=int,

components/clp-package-utils/clp_package_utils/scripts/native/archive_manager.py

@@ -97,6 +97,11 @@ def main(argv: typing.List[str]) -> int:
         default=str(default_config_file_path),
         help="CLP configuration file.",
     )
+    args_parser.add_argument(
+        "--dataset",
+        default=CLP_DEFAULT_DATASET_NAME,
+        help="The name of the log category.",
+    )
 
     # Top-level commands
     subparsers: argparse._SubParsersAction[argparse.ArgumentParser] = args_parser.add_subparsers(
@@ -196,7 +201,7 @@ def main(argv: typing.List[str]) -> int:
             archives_dir,
             database_config,
             storage_engine,
-            CLP_DEFAULT_DATASET_NAME,
+            parsed_args.dataset,
             parsed_args.begin_ts,
             parsed_args.end_ts,
         )
@@ -208,7 +213,7 @@ def main(argv: typing.List[str]) -> int:
                 archives_dir,
                 database_config,
                 storage_engine,
-                CLP_DEFAULT_DATASET_NAME,
+                parsed_args.dataset,
                 delete_handler,
                 parsed_args.dry_run,
             )
@@ -220,7 +225,7 @@ def main(argv: typing.List[str]) -> int:
                 archives_dir,
                 database_config,
                 storage_engine,
-                CLP_DEFAULT_DATASET_NAME,
+                parsed_args.dataset,
                 delete_handler,
                 parsed_args.dry_run,
             )

components/clp-package-utils/clp_package_utils/scripts/native/compress.py

@@ -10,7 +10,11 @@
 
 import brotli
 import msgpack
-from clp_py_utils.clp_config import CLPConfig, COMPRESSION_JOBS_TABLE_NAME
+from clp_py_utils.clp_config import (
+    CLP_DEFAULT_DATASET_NAME,
+    CLPConfig,
+    COMPRESSION_JOBS_TABLE_NAME,
+)
 from clp_py_utils.pretty_size import pretty_size
 from clp_py_utils.s3_utils import parse_s3_url
 from clp_py_utils.sql_adapter import SQL_Adapter
@@ -140,6 +144,7 @@ def _generate_clp_io_config(
         if len(logs_to_compress) == 0:
             raise ValueError(f"No input paths given.")
         return FsInputConfig(
+            dataset=parsed_args.dataset,
             paths_to_compress=logs_to_compress,
             timestamp_key=parsed_args.timestamp_key,
             path_prefix_to_remove=str(CONTAINER_INPUT_LOGS_ROOT_DIR),
@@ -154,6 +159,7 @@ def _generate_clp_io_config(
         region_code, bucket_name, key_prefix = parse_s3_url(s3_url)
         aws_authentication = clp_config.logs_input.aws_authentication
         return S3InputConfig(
+            dataset=parsed_args.dataset,
             region_code=region_code,
             bucket=bucket_name,
             key_prefix=key_prefix,
@@ -190,6 +196,11 @@ def main(argv):
         default=str(default_config_file_path),
         help="CLP package configuration file.",
     )
+    args_parser.add_argument(
+        "--dataset",
+        default=CLP_DEFAULT_DATASET_NAME,
+        help="The name of the log category.",
+    )
     args_parser.add_argument(
         "-f",
         "--logs-list",

components/clp-package-utils/clp_package_utils/scripts/native/decompress.py

@@ -14,6 +14,7 @@
     CLPConfig,
     Database,
     FILES_TABLE_SUFFIX,
+    StorageEngine,
 )
 from clp_py_utils.sql_adapter import SQL_Adapter
 from job_orchestration.scheduler.constants import QueryJobStatus, QueryJobType
@@ -40,9 +41,13 @@
 logger = logging.getLogger(__file__)
 
 
-def get_orig_file_id(db_config: Database, path: str) -> Optional[str]:
+def get_orig_file_id(
+    db_config: Database, storage_engine: StorageEngine, dataset: str, path: str
+) -> Optional[str]:
     """
     :param db_config:
+    :param storage_engine:
+    :param dataset:
     :param path: Path of the original file.
     :return: The ID of an original file which has the given path, or None if no such file exists.
     NOTE: Multiple original files may have the same path in which case this method returns the ID of
@@ -51,6 +56,9 @@ def get_orig_file_id(db_config: Database, path: str) -> Optional[str]:
     sql_adapter = SQL_Adapter(db_config)
     clp_db_connection_params = db_config.get_clp_connection_params_and_type(True)
     table_prefix = clp_db_connection_params["table_prefix"]
+    if StorageEngine.CLP_S == storage_engine:
+        table_prefix = f"{table_prefix}{dataset}_"
+
     with closing(sql_adapter.create_connection(True)) as db_conn, closing(
         db_conn.cursor(dictionary=True)
     ) as db_cursor:
@@ -128,7 +136,12 @@ def handle_extract_stream_cmd(
             orig_file_id = parsed_args.orig_file_id
         else:
             orig_file_path = parsed_args.orig_file_path
-            orig_file_id = get_orig_file_id(clp_config.database, orig_file_path)
+            orig_file_id = get_orig_file_id(
+                clp_config.database,
+                clp_config.package.storage_engine,
+                CLP_DEFAULT_DATASET_NAME,
+                orig_file_path,
+            )
             if orig_file_id is None:
                 logger.error(f"Cannot find orig_file_id corresponding to '{orig_file_path}'.")
                 return -1
@@ -140,7 +153,7 @@ def handle_extract_stream_cmd(
     elif EXTRACT_JSON_CMD == command:
         job_type = QueryJobType.EXTRACT_JSON
         job_config = ExtractJsonJobConfig(
-            dataset=CLP_DEFAULT_DATASET_NAME,
+            dataset=parsed_args.dataset,
             archive_id=parsed_args.archive_id,
             target_chunk_size=parsed_args.target_chunk_size,
         )
@@ -299,6 +312,12 @@ def main(argv):
     # JSON extraction command parser
     json_extraction_parser = command_args_parser.add_parser(EXTRACT_JSON_CMD)
     json_extraction_parser.add_argument("archive_id", type=str, help="Archive ID")
+    json_extraction_parser.add_argument(
+        "--dataset",
+        type=str,
+        default=CLP_DEFAULT_DATASET_NAME,
+        help="The name of the log category.",
+    )
     json_extraction_parser.add_argument(
         "--target-chunk-size", type=int, help="Target chunk size (B)."
     )

components/clp-package-utils/clp_package_utils/scripts/native/search.py

@@ -10,7 +10,11 @@
 
 import msgpack
 import pymongo
-from clp_py_utils.clp_config import Database, ResultsCache
+from clp_py_utils.clp_config import (
+    CLP_DEFAULT_DATASET_NAME,
+    Database,
+    ResultsCache,
+)
 from clp_py_utils.sql_adapter import SQL_Adapter
 from job_orchestration.scheduler.constants import QueryJobStatus, QueryJobType
 from job_orchestration.scheduler.job_config import AggregationConfig, SearchJobConfig
@@ -32,6 +36,7 @@
 def create_and_monitor_job_in_db(
     db_config: Database,
     results_cache: ResultsCache,
+    dataset: str,
     wildcard_query: str,
     tags: str | None,
     begin_timestamp: int | None,
@@ -43,6 +48,7 @@ def create_and_monitor_job_in_db(
     count_by_time_bucket_size: int | None,
 ):
     search_config = SearchJobConfig(
+        dataset=dataset,
         query_string=wildcard_query,
         begin_timestamp=begin_timestamp,
         end_timestamp=end_timestamp,
@@ -113,6 +119,7 @@ async def worker_connection_handler(reader: asyncio.StreamReader, writer: asynci
 async def do_search_without_aggregation(
     db_config: Database,
     results_cache: ResultsCache,
+    dataset: str,
     wildcard_query: str,
     tags: str | None,
     begin_timestamp: int | None,
@@ -147,6 +154,7 @@ async def do_search_without_aggregation(
             create_and_monitor_job_in_db,
             db_config,
             results_cache,
+            dataset,
             wildcard_query,
             tags,
             begin_timestamp,
@@ -184,6 +192,7 @@ async def do_search_without_aggregation(
 async def do_search(
     db_config: Database,
     results_cache: ResultsCache,
+    dataset: str,
     wildcard_query: str,
     tags: str | None,
     begin_timestamp: int | None,
@@ -198,6 +207,7 @@ async def do_search(
         await do_search_without_aggregation(
             db_config,
             results_cache,
+            dataset,
             wildcard_query,
             tags,
             begin_timestamp,
@@ -211,6 +221,7 @@ async def do_search(
             create_and_monitor_job_in_db,
             db_config,
             results_cache,
+            dataset,
             wildcard_query,
             tags,
             begin_timestamp,
@@ -229,6 +240,11 @@ def main(argv):
 
     args_parser = argparse.ArgumentParser(description="Searches the compressed logs.")
     args_parser.add_argument("--config", "-c", required=True, help="CLP configuration file.")
+    args_parser.add_argument(
+        "--dataset",
+        default=CLP_DEFAULT_DATASET_NAME,
+        help="The name of the log category.",
+    )
     args_parser.add_argument("wildcard_query", help="Wildcard query.")
     args_parser.add_argument(
         "-t", "--tags", help="Comma-separated list of tags of archives to search."
@@ -286,6 +302,7 @@ def main(argv):
             do_search(
                 clp_config.database,
                 clp_config.results_cache,
+                parsed_args.dataset,
                 parsed_args.wildcard_query,
                 parsed_args.tags,
                 parsed_args.begin_time,

components/clp-package-utils/clp_package_utils/scripts/search.py

@@ -7,7 +7,11 @@
 import uuid
 
 import yaml
-from clp_py_utils.clp_config import StorageEngine, StorageType
+from clp_py_utils.clp_config import (
+    CLP_DEFAULT_DATASET_NAME,
+    StorageEngine,
+    StorageType,
+)
 
 from clp_package_utils.general import (
     CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH,
@@ -35,6 +39,11 @@ def main(argv):
         default=str(default_config_file_path),
         help="CLP package configuration file.",
     )
+    args_parser.add_argument(
+        "--dataset",
+        default=CLP_DEFAULT_DATASET_NAME,
+        help="The name of the log category.",
+    )
     args_parser.add_argument("wildcard_query", help="Wildcard query.")
     args_parser.add_argument(
         "-t", "--tags", help="Comma-separated list of tags of archives to search."
@@ -104,6 +113,7 @@ def main(argv):
         "python3",
         "-m", "clp_package_utils.scripts.native.search",
         "--config", str(generated_config_path_on_container),
+        "--dataset", str(parsed_args.dataset),
         parsed_args.wildcard_query,
     ]
     # fmt: on