Merge branch 'main' into modernize-liblzma-install

Author: Bill-hbrhbr

components/clp-package-utils/clp_package_utils/general.py

@@ -16,6 +16,7 @@
     CLP_DEFAULT_CREDENTIALS_FILE_PATH,
     CLPConfig,
     DB_COMPONENT_NAME,
+    QueryEngine,
     QUEUE_COMPONENT_NAME,
     REDIS_COMPONENT_NAME,
     REDUCER_COMPONENT_NAME,
@@ -96,6 +97,13 @@ def __init__(self, clp_home: pathlib.Path, docker_clp_home: pathlib.Path):
         self.aws_config_dir: typing.Optional[DockerMount] = None
 
 
+def _validate_data_directory(data_dir: pathlib.Path, component_name: str) -> None:
+    try:
+        validate_path_could_be_dir(data_dir)
+    except ValueError as ex:
+        raise ValueError(f"{component_name} data directory is invalid: {ex}")
+
+
 def get_clp_home():
     # Determine CLP_HOME from an environment variable or this script's path
     clp_home = None
@@ -175,6 +183,13 @@ def is_container_exited(container_name):
     return False
 
 
+def validate_log_directory(logs_dir: pathlib.Path, component_name: str) -> None:
+    try:
+        validate_path_could_be_dir(logs_dir)
+    except ValueError as ex:
+        raise ValueError(f"{component_name} logs directory is invalid: {ex}")
+
+
 def validate_port(port_name: str, hostname: str, port: int):
     try:
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -431,40 +446,23 @@ def validate_and_load_redis_credentials_file(
 
 
 def validate_db_config(clp_config: CLPConfig, data_dir: pathlib.Path, logs_dir: pathlib.Path):
-    try:
-        validate_path_could_be_dir(data_dir)
-    except ValueError as ex:
-        raise ValueError(f"{DB_COMPONENT_NAME} data directory is invalid: {ex}")
-
-    try:
-        validate_path_could_be_dir(logs_dir)
-    except ValueError as ex:
-        raise ValueError(f"{DB_COMPONENT_NAME} logs directory is invalid: {ex}")
+    _validate_data_directory(data_dir, DB_COMPONENT_NAME)
+    validate_log_directory(logs_dir, DB_COMPONENT_NAME)
 
     validate_port(f"{DB_COMPONENT_NAME}.port", clp_config.database.host, clp_config.database.port)
 
 
 def validate_queue_config(clp_config: CLPConfig, logs_dir: pathlib.Path):
-    try:
-        validate_path_could_be_dir(logs_dir)
-    except ValueError as ex:
-        raise ValueError(f"{QUEUE_COMPONENT_NAME} logs directory is invalid: {ex}")
+    validate_log_directory(logs_dir, QUEUE_COMPONENT_NAME)
 
     validate_port(f"{QUEUE_COMPONENT_NAME}.port", clp_config.queue.host, clp_config.queue.port)
 
 
 def validate_redis_config(
     clp_config: CLPConfig, data_dir: pathlib.Path, logs_dir: pathlib.Path, base_config: pathlib.Path
 ):
-    try:
-        validate_path_could_be_dir(data_dir)
-    except ValueError as ex:
-        raise ValueError(f"{REDIS_COMPONENT_NAME} data directory is invalid {ex}")
-
-    try:
-        validate_path_could_be_dir(logs_dir)
-    except ValueError as ex:
-        raise ValueError(f"{REDIS_COMPONENT_NAME} logs directory is invalid: {ex}")
+    _validate_data_directory(data_dir, REDIS_COMPONENT_NAME)
+    validate_log_directory(logs_dir, REDIS_COMPONENT_NAME)
 
     if not base_config.exists():
         raise ValueError(
@@ -475,10 +473,7 @@ def validate_redis_config(
 
 
 def validate_reducer_config(clp_config: CLPConfig, logs_dir: pathlib.Path, num_workers: int):
-    try:
-        validate_path_could_be_dir(logs_dir)
-    except ValueError as ex:
-        raise ValueError(f"{REDUCER_COMPONENT_NAME} logs directory is invalid: {ex}")
+    validate_log_directory(logs_dir, REDUCER_COMPONENT_NAME)
 
     for i in range(0, num_workers):
         validate_port(
@@ -491,15 +486,8 @@ def validate_reducer_config(clp_config: CLPConfig, logs_dir: pathlib.Path, num_w
 def validate_results_cache_config(
     clp_config: CLPConfig, data_dir: pathlib.Path, logs_dir: pathlib.Path
 ):
-    try:
-        validate_path_could_be_dir(data_dir)
-    except ValueError as ex:
-        raise ValueError(f"{RESULTS_CACHE_COMPONENT_NAME} data directory is invalid: {ex}")
-
-    try:
-        validate_path_could_be_dir(logs_dir)
-    except ValueError as ex:
-        raise ValueError(f"{RESULTS_CACHE_COMPONENT_NAME} logs directory is invalid: {ex}")
+    _validate_data_directory(data_dir, RESULTS_CACHE_COMPONENT_NAME)
+    validate_log_directory(logs_dir, RESULTS_CACHE_COMPONENT_NAME)
 
     validate_port(
         f"{RESULTS_CACHE_COMPONENT_NAME}.port",
@@ -508,8 +496,11 @@ def validate_results_cache_config(
     )
 
 
-def validate_worker_config(clp_config: CLPConfig):
+def validate_logs_input_config(clp_config: CLPConfig) -> None:
     clp_config.validate_logs_input_config()
+
+
+def validate_output_storage_config(clp_config: CLPConfig) -> None:
     clp_config.validate_archive_output_config()
     clp_config.validate_stream_output_config()
 
@@ -590,3 +581,21 @@ def validate_dataset_name(clp_table_prefix: str, dataset_name: str) -> None:
             f"Invalid dataset name: `{dataset_name}`. Names can only be a maximum of"
             f" {dataset_name_max_len} characters long."
         )
+
+
+def validate_retention_config(clp_config: CLPConfig) -> None:
+    clp_query_engine = clp_config.package.query_engine
+    if is_retention_period_configured(clp_config) and clp_query_engine == QueryEngine.PRESTO:
+        raise ValueError(
+            f"Retention control is not supported with query_engine `{clp_query_engine}`"
+        )
+
+
+def is_retention_period_configured(clp_config: CLPConfig) -> bool:
+    if clp_config.archive_output.retention_period is not None:
+        return True
+
+    if clp_config.results_cache.retention_period is not None:
+        return True
+
+    return False

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

@@ -26,15 +26,13 @@
 )
 
 # Command/Argument Constants
-from clp_package_utils.scripts.native.archive_manager import (
-    BEGIN_TS_ARG,
-    DEL_BY_FILTER_SUBCOMMAND,
-    DEL_BY_IDS_SUBCOMMAND,
-    DEL_COMMAND,
-    DRY_RUN_ARG,
-    END_TS_ARG,
-    FIND_COMMAND,
-)
+FIND_COMMAND: typing.Final[str] = "find"
+DEL_COMMAND: typing.Final[str] = "del"
+DEL_BY_IDS_SUBCOMMAND: typing.Final[str] = "by-ids"
+DEL_BY_FILTER_SUBCOMMAND: typing.Final[str] = "by-filter"
+BEGIN_TS_ARG: typing.Final[str] = "--begin-ts"
+END_TS_ARG: typing.Final[str] = "--end-ts"
+DRY_RUN_ARG: typing.Final[str] = "--dry-run"
 
 logger: logging.Logger = logging.getLogger(__file__)
 
@@ -66,6 +64,12 @@ def main(argv: typing.List[str]) -> int:
         default=str(default_config_file_path),
         help="CLP package configuration file.",
     )
+    args_parser.add_argument(
+        "--verbose",
+        "-v",
+        action="store_true",
+        help="Enable debug logging.",
+    )
     args_parser.add_argument(
         "--dataset",
         type=str,
@@ -150,6 +154,10 @@ def main(argv: typing.List[str]) -> int:
     )
 
     parsed_args: argparse.Namespace = args_parser.parse_args(argv[1:])
+    if parsed_args.verbose:
+        logger.setLevel(logging.DEBUG)
+    else:
+        logger.setLevel(logging.INFO)
 
     begin_timestamp: typing.Optional[int]
     end_timestamp: typing.Optional[int]
@@ -171,7 +179,7 @@ def main(argv: typing.List[str]) -> int:
 
     storage_type: StorageType = clp_config.archive_output.storage.type
     if StorageType.FS != storage_type:
-        logger.error(f"Archive deletion is not supported for storage type: {storage_type}.")
+        logger.error(f"Archive manager is not supported for storage type: {storage_type}.")
         return -1
 
     storage_engine: StorageEngine = clp_config.package.storage_engine
@@ -226,6 +234,9 @@ def main(argv: typing.List[str]) -> int:
     if dataset is not None:
         archive_manager_cmd.append("--dataset")
         archive_manager_cmd.append(dataset)
+    if parsed_args.verbose:
+        archive_manager_cmd.append("--verbose")
+
     archive_manager_cmd.append(subcommand)
 
     # Add subcommand-specific arguments
@@ -251,15 +262,20 @@ def main(argv: typing.List[str]) -> int:
             archive_manager_cmd.extend([END_TS_ARG, str(end_timestamp)])
     else:
         logger.error(f"Unsupported subcommand: `{subcommand}`.")
+        return -1
 
     cmd: typing.List[str] = container_start_cmd + archive_manager_cmd
 
-    subprocess.run(cmd, check=True)
+    proc = subprocess.run(cmd)
+    ret_code = proc.returncode
+    if 0 != ret_code:
+        logger.error("Archive manager failed.")
+        logger.debug(f"Docker command failed: {' '.join(cmd)}")
 
     # Remove generated files
     generated_config_path_on_host.unlink()
 
-    return 0
+    return ret_code
 
 
 if "__main__" == __name__:

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

@@ -181,23 +181,13 @@ def handle_extract_stream_cmd(
         return -1
 
     job_command = parsed_args.command
+    if EXTRACT_IR_CMD == job_command and StorageEngine.CLP != storage_engine:
+        logger.error(f"IR extraction is not supported for storage engine `{storage_engine}`.")
+        return -1
     if EXTRACT_JSON_CMD == job_command and StorageEngine.CLP_S != storage_engine:
         logger.error(f"JSON extraction is not supported for storage engine `{storage_engine}`.")
         return -1
 
-    dataset = parsed_args.dataset
-    if StorageEngine.CLP_S == storage_engine:
-        dataset = CLP_DEFAULT_DATASET_NAME if dataset is None else dataset
-        try:
-            clp_db_connection_params = clp_config.database.get_clp_connection_params_and_type(True)
-            validate_dataset_name(clp_db_connection_params["table_prefix"], dataset)
-        except Exception as e:
-            logger.error(e)
-            return -1
-    elif dataset is not None:
-        logger.error(f"Dataset selection is not supported for storage engine: {storage_engine}.")
-        return -1
-
     container_name = generate_container_name(str(JobType.IR_EXTRACTION))
     container_clp_config, mounts = generate_container_config(clp_config, clp_home)
     generated_config_path_on_container, generated_config_path_on_host = dump_container_config(
@@ -229,6 +219,15 @@ def handle_extract_stream_cmd(
             extract_cmd.append("--target-uncompressed-size")
             extract_cmd.append(str(parsed_args.target_uncompressed_size))
     elif EXTRACT_JSON_CMD == job_command:
+        dataset = parsed_args.dataset
+        dataset = CLP_DEFAULT_DATASET_NAME if dataset is None else dataset
+        try:
+            clp_db_connection_params = clp_config.database.get_clp_connection_params_and_type(True)
+            validate_dataset_name(clp_db_connection_params["table_prefix"], dataset)
+        except Exception as e:
+            logger.error(e)
+            return -1
+
         extract_cmd.append(str(parsed_args.archive_id))
         if dataset is not None:
             extract_cmd.append("--dataset")