feat(clp-package)!: Refactor input log config so that S3 source locations are specified in the config file rather than through the CLI.

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

@@ -20,6 +20,7 @@
     REDIS_COMPONENT_NAME,
     REDUCER_COMPONENT_NAME,
     RESULTS_CACHE_COMPONENT_NAME,
+    StorageType,
     WEBUI_COMPONENT_NAME,
     WorkerConfig,
 )
@@ -216,13 +217,14 @@ def generate_container_config(
 
     docker_mounts = CLPDockerMounts(clp_home, CONTAINER_CLP_HOME)
 
-    input_logs_dir = clp_config.input_logs_directory.resolve()
-    container_clp_config.input_logs_directory = (
-        CONTAINER_INPUT_LOGS_ROOT_DIR / input_logs_dir.relative_to(input_logs_dir.anchor)
-    )
-    docker_mounts.input_logs_dir = DockerMount(
-        DockerMountType.BIND, input_logs_dir, container_clp_config.input_logs_directory, True
-    )
+    if StorageType.FS == clp_config.logs_input.type:
+        input_logs_dir = clp_config.logs_input.directory.resolve()
+        container_clp_config.logs_input.directory = (
+            CONTAINER_INPUT_LOGS_ROOT_DIR / input_logs_dir.relative_to(input_logs_dir.anchor)
+        )
+        docker_mounts.input_logs_dir = DockerMount(
+            DockerMountType.BIND, input_logs_dir, container_clp_config.logs_input.directory, True
+        )
 
     container_clp_config.data_directory = CONTAINER_CLP_HOME / "var" / "data"
     if not is_path_already_mounted(
@@ -494,7 +496,7 @@ def validate_results_cache_config(
 
 
 def validate_worker_config(clp_config: CLPConfig):
-    clp_config.validate_input_logs_dir()
+    clp_config.validate_logs_input_config()
     clp_config.validate_archive_output_config()
     clp_config.validate_stream_output_dir()
 

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

@@ -1,11 +1,10 @@
 import argparse
-import configparser
 import logging
 import pathlib
 import subprocess
 import sys
 import uuid
-from typing import List, Tuple
+from typing import List
 
 from clp_py_utils.clp_config import CLPConfig, StorageEngine
 from job_orchestration.scheduler.job_config import InputType
@@ -26,46 +25,11 @@
 logger = logging.getLogger(__file__)
 
 
-def _parse_aws_credentials_file(credentials_file_path: pathlib.Path, user: str) -> Tuple[str, str]:
-    """
-    Parses the `aws_access_key_id` and `aws_secret_access_key` of `user` from the given
-    credentials_file_path.
-    :param credentials_file_path:
-    :param user:
-    :return: A tuple of (aws_access_key_id, aws_secret_access_key)
-    :raises: ValueError if the file doesn't exist, or doesn't contain valid aws credentials.
-    """
-
-    if not credentials_file_path.exists():
-        raise ValueError(f"'{credentials_file_path}' doesn't exist.")
-
-    config_reader = configparser.ConfigParser()
-    config_reader.read(credentials_file_path)
-
-    if not config_reader.has_section(user):
-        raise ValueError(f"User '{user}' doesn't exist.")
-
-    user_credentials = config_reader[user]
-    if "aws_session_token" in user_credentials:
-        raise ValueError(f"Session tokens (short-term credentials) are not supported.")
-
-    aws_access_key_id = user_credentials.get("aws_access_key_id")
-    aws_secret_access_key = user_credentials.get("aws_secret_access_key")
-
-    if aws_access_key_id is None or aws_secret_access_key is None:
-        raise ValueError(
-            "The credentials file must contain both aws_access_key_id and aws_secret_access_key."
-        )
-
-    return aws_access_key_id, aws_secret_access_key
-
-
 def _generate_logs_list(
+    input_type: InputType,
     container_logs_list_path: pathlib.Path,
     parsed_args: argparse.Namespace,
 ) -> None:
-    input_type = parsed_args.input_type
-
     if InputType.FS == input_type:
         host_logs_list_path = parsed_args.path_list
         with open(container_logs_list_path, "w") as container_logs_list_file:
@@ -91,23 +55,23 @@ def _generate_logs_list(
 
     elif InputType.S3 == input_type:
         with open(container_logs_list_path, "w") as container_logs_list_file:
-            container_logs_list_file.write(f"{parsed_args.url}\n")
+            container_logs_list_file.write(f"{parsed_args.paths[0]}\n")
 
     else:
         raise ValueError(f"Unsupported input type: {input_type}.")
 
 
 def _generate_compress_cmd(
-    parsed_args: argparse.Namespace, config_path: pathlib.Path, logs_list_path: pathlib.Path
+    parsed_args: argparse.Namespace,
+    config_path: pathlib.Path,
+    logs_list_path: pathlib.Path,
 ) -> List[str]:
-    input_type = parsed_args.input_type
 
     # fmt: off
     compress_cmd = [
         "python3",
         "-m", "clp_package_utils.scripts.native.compress",
         "--config", str(config_path),
-        input_type,
     ]
     # fmt: on
     if parsed_args.timestamp_key is not None:
@@ -119,43 +83,12 @@ def _generate_compress_cmd(
     if parsed_args.no_progress_reporting is True:
         compress_cmd.append("--no-progress-reporting")
 
-    if InputType.FS == input_type:
-        pass
-    elif InputType.S3 == input_type:
-        aws_access_key_id = parsed_args.aws_access_key_id
-        aws_secret_access_key = parsed_args.aws_secret_access_key
-        if parsed_args.aws_credentials_file:
-            default_credentials_user = "default"
-            aws_access_key_id, aws_secret_access_key = _parse_aws_credentials_file(
-                pathlib.Path(parsed_args.aws_credentials_file), default_credentials_user
-            )
-        if bool(aws_access_key_id) and bool(aws_secret_access_key):
-            compress_cmd.append("--aws-access-key-id")
-            compress_cmd.append(aws_access_key_id)
-            compress_cmd.append("--aws-secret-access-key")
-            compress_cmd.append(aws_secret_access_key)
-    else:
-        raise ValueError(f"Unsupported input type: {input_type}.")
-
     compress_cmd.append("--logs-list")
     compress_cmd.append(str(logs_list_path))
 
     return compress_cmd
 
 
-def _add_common_arguments(args_parser: argparse.ArgumentParser) -> None:
-    args_parser.add_argument(
-        "--timestamp-key",
-        help="The path (e.g. x.y) for the field containing the log event's timestamp.",
-    )
-    args_parser.add_argument(
-        "-t", "--tags", help="A comma-separated list of tags to apply to the compressed archives."
-    )
-    args_parser.add_argument(
-        "--no-progress-reporting", action="store_true", help="Disables progress reporting."
-    )
-
-
 def _validate_fs_input_args(
     parsed_args: argparse.Namespace,
     args_parser: argparse.ArgumentParser,
@@ -170,33 +103,19 @@ def _validate_fs_input_args(
 
 
 def _validate_s3_input_args(
-    parsed_args: argparse.Namespace, args_parser: argparse.ArgumentParser, clp_config: CLPConfig
+    parsed_args: argparse.Namespace,
+    args_parser: argparse.ArgumentParser,
+    storage_engine: StorageEngine,
 ) -> None:
-    if StorageEngine.CLP_S != clp_config.package.storage_engine:
+    if StorageEngine.CLP_S != storage_engine:
         args_parser.error(
             f"Input type {InputType.S3} is only supported for the storage engine"
             f" {StorageEngine.CLP_S}."
         )
-
-    # Validate aws credentials were specified using only one method
-    aws_credential_file = parsed_args.aws_credentials_file
-    aws_access_key_id = parsed_args.aws_access_key_id
-    aws_secret_access_key = parsed_args.aws_secret_access_key
-    if aws_credential_file is not None:
-        if not pathlib.Path(aws_credential_file).exists():
-            args_parser.error(f"AWS credentials file '{aws_credential_file}' doesn't exist.")
-
-        if aws_access_key_id is not None or aws_secret_access_key is not None:
-            args_parser.error(
-                "aws_credentials_file cannot be specified together with aws_access_key_id or"
-                " aws_secret_access_key."
-            )
-
-    else:
-        if not bool(aws_access_key_id):
-            args_parser.error("aws_access_key_id not specified or empty")
-        if not bool(aws_secret_access_key):
-            args_parser.error("aws_secret_access_key not specified or empty")
+    if len(parsed_args.paths) != 1:
+        args_parser.error(f"Only one key prefix can be specified for input type {InputType.S3}.")
+    if parsed_args.path_list is not None:
+        args_parser.error(f"Path list file is unsupported for input type {InputType.S3}.")
 
 
 def main(argv):
@@ -212,26 +131,19 @@ def main(argv):
         default=str(default_config_file_path),
         help="CLP package configuration file.",
     )
-    input_type_args_parser = args_parser.add_subparsers(dest="input_type")
-
-    fs_compressor_parser = input_type_args_parser.add_parser(InputType.FS)
-    _add_common_arguments(fs_compressor_parser)
-    fs_compressor_parser.add_argument("paths", metavar="PATH", nargs="*", help="Paths to compress.")
-    fs_compressor_parser.add_argument(
-        "-f", "--path-list", dest="path_list", help="A file listing all paths to compress."
+    args_parser.add_argument(
+        "--timestamp-key",
+        help="The path (e.g. x.y) for the field containing the log event's timestamp.",
     )
-
-    s3_compressor_parser = input_type_args_parser.add_parser(InputType.S3)
-    _add_common_arguments(s3_compressor_parser)
-    s3_compressor_parser.add_argument("url", metavar="URL", help="URL of objects to be compressed")
-    s3_compressor_parser.add_argument(
-        "--aws-access-key-id", type=str, default=None, help="AWS access key ID."
+    args_parser.add_argument(
+        "-t", "--tags", help="A comma-separated list of tags to apply to the compressed archives."
     )
-    s3_compressor_parser.add_argument(
-        "--aws-secret-access-key", type=str, default=None, help="AWS secret access key."
+    args_parser.add_argument(
+        "--no-progress-reporting", action="store_true", help="Disables progress reporting."
     )
-    s3_compressor_parser.add_argument(
-        "--aws-credentials-file", type=str, default=None, help="Path to AWS credentials file."
+    args_parser.add_argument("paths", metavar="PATH", nargs="*", help="Paths to compress.")
+    args_parser.add_argument(
+        "-f", "--path-list", dest="path_list", help="A file listing all paths to compress."
     )
 
     parsed_args = args_parser.parse_args(argv[1:])
@@ -248,11 +160,11 @@ def main(argv):
         logger.exception("Failed to load config.")
         return -1
 
-    input_type = parsed_args.input_type
+    input_type = clp_config.logs_input.type
     if InputType.FS == input_type:
         _validate_fs_input_args(parsed_args, args_parser)
     elif InputType.S3 == input_type:
-        _validate_s3_input_args(parsed_args, args_parser, clp_config)
+        _validate_s3_input_args(parsed_args, args_parser, clp_config.package.storage_engine)
     else:
         raise ValueError(f"Unsupported input type: {input_type}.")
 
@@ -263,7 +175,9 @@ def main(argv):
         container_clp_config, clp_config, container_name
     )
 
-    necessary_mounts = [mounts.clp_home, mounts.input_logs_dir, mounts.data_dir, mounts.logs_dir]
+    necessary_mounts = [mounts.clp_home, mounts.data_dir, mounts.logs_dir]
+    if InputType.FS == input_type:
+        necessary_mounts.append(mounts.input_logs_dir)
 
     # Write compression logs to a file
     while True:
@@ -276,7 +190,7 @@ def main(argv):
         if not container_logs_list_path.exists():
             break
 
-    _generate_logs_list(container_logs_list_path, parsed_args)
+    _generate_logs_list(clp_config.logs_input.type, container_logs_list_path, parsed_args)
 
     container_start_cmd = generate_container_start_cmd(
         container_name, necessary_mounts, clp_config.execution_container