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,45 +25,12 @@
 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(
     container_logs_list_path: pathlib.Path,
     parsed_args: argparse.Namespace,
+    clp_config: CLPConfig,
 ) -> None:
-    input_type = parsed_args.input_type
+    input_type = clp_config.logs_input.type
 
     if InputType.FS == input_type:
         host_logs_list_path = parsed_args.path_list
@@ -91,23 +57,25 @@ 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,
+    clp_config: CLPConfig,
 ) -> List[str]:
-    input_type = parsed_args.input_type
+    input_type = clp_config.logs_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 +87,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,
@@ -177,26 +114,10 @@ def _validate_s3_input_args(
             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 not supported for input type {InputType.S3}.")
 
 
 def main(argv):
@@ -212,26 +133,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,7 +162,7 @@ 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:
@@ -263,7 +177,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,13 +192,13 @@ def main(argv):
         if not container_logs_list_path.exists():
             break
 
-    _generate_logs_list(container_logs_list_path, parsed_args)
+    _generate_logs_list(container_logs_list_path, parsed_args, clp_config)
 
     container_start_cmd = generate_container_start_cmd(
         container_name, necessary_mounts, clp_config.execution_container
     )
     compress_cmd = _generate_compress_cmd(
-        parsed_args, generated_config_path_on_container, logs_list_path_on_container
+        parsed_args, generated_config_path_on_container, logs_list_path_on_container, clp_config
     )
     cmd = container_start_cmd + compress_cmd
     subprocess.run(cmd, check=True)

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

@@ -10,9 +10,8 @@
 
 import brotli
 import msgpack
-from clp_py_utils.clp_config import COMPRESSION_JOBS_TABLE_NAME, S3Credentials
+from clp_py_utils.clp_config import CLPConfig, COMPRESSION_JOBS_TABLE_NAME, S3Credentials
 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
 from job_orchestration.scheduler.constants import (
     CompressionJobCompletionStatus,
@@ -128,9 +127,9 @@ def handle_job(sql_adapter: SQL_Adapter, clp_io_config: ClpIoConfig, no_progress
 
 
 def _generate_clp_io_config(
-    logs_to_compress: List[str], parsed_args: argparse.Namespace
+    logs_to_compress: List[str], parsed_args: argparse.Namespace, clp_config: CLPConfig
 ) -> typing.Union[S3InputConfig, FsInputConfig]:
-    input_type = parsed_args.input_type
+    input_type = clp_config.logs_input.type
 
     if InputType.FS == input_type:
         return FsInputConfig(
@@ -140,18 +139,14 @@ def _generate_clp_io_config(
         )
     elif InputType.S3 == input_type:
         if len(logs_to_compress) != 1:
-            ValueError(f"Too many URLs: {len(logs_to_compress)} > 1")
+            raise ValueError(f"Too many prefixes: {len(logs_to_compress)} > 1")
 
-        s3_url = logs_to_compress[0]
-        region_code, bucket_name, key_prefix = parse_s3_url(s3_url)
+        s3_config = clp_config.logs_input.s3_config
         return S3InputConfig(
-            region_code=region_code,
-            bucket=bucket_name,
-            key_prefix=key_prefix,
-            credentials=S3Credentials(
-                access_key_id=parsed_args.aws_access_key_id,
-                secret_access_key=parsed_args.aws_secret_access_key,
-            ),
+            region_code=s3_config.region_code,
+            bucket=s3_config.bucket,
+            key_prefix=s3_config.key_prefix + logs_to_compress[0],
+            credentials=s3_config.credentials,
             timestamp_key=parsed_args.timestamp_key,
         )
     else:
@@ -175,7 +170,18 @@ def _get_logs_to_compress(logs_list_path: pathlib.Path) -> List[str]:
     return logs_to_compress
 
 
-def _add_common_arguments(args_parser: argparse.ArgumentParser) -> None:
+def main(argv):
+    clp_home = get_clp_home()
+    default_config_file_path = clp_home / CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH
+    args_parser = argparse.ArgumentParser(description="Compresses logs")
+
+    # Package-level config option
+    args_parser.add_argument(
+        "--config",
+        "-c",
+        default=str(default_config_file_path),
+        help="CLP package configuration file.",
+    )
     args_parser.add_argument(
         "-f",
         "--logs-list",
@@ -193,41 +199,13 @@ def _add_common_arguments(args_parser: argparse.ArgumentParser) -> None:
     args_parser.add_argument(
         "-t", "--tags", help="A comma-separated list of tags to apply to the compressed archives."
     )
-
-
-def main(argv):
-    clp_home = get_clp_home()
-    default_config_file_path = clp_home / CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH
-    args_parser = argparse.ArgumentParser(description="Compresses logs")
-
-    # Package-level config option
-    args_parser.add_argument(
-        "--config",
-        "-c",
-        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)
-
-    s3_compressor_parser = input_type_args_parser.add_parser(InputType.S3)
-    _add_common_arguments(s3_compressor_parser)
-    s3_compressor_parser.add_argument(
-        "--aws-access-key-id", type=str, default=None, help="AWS access key ID."
-    )
-    s3_compressor_parser.add_argument(
-        "--aws-secret-access-key", type=str, default=None, help="AWS secret access key."
-    )
-
     parsed_args = args_parser.parse_args(argv[1:])
 
     # Validate and load config file
     try:
         config_file_path = pathlib.Path(parsed_args.config)
         clp_config = load_config_file(config_file_path, default_config_file_path, clp_home)
-        clp_config.validate_input_logs_dir()
+        clp_config.validate_logs_input_config()
         clp_config.validate_logs_dir()
     except:
         logger.exception("Failed to load config.")
@@ -238,7 +216,7 @@ def main(argv):
 
     logs_to_compress = _get_logs_to_compress(pathlib.Path(parsed_args.logs_list).resolve())
 
-    clp_input_config = _generate_clp_io_config(logs_to_compress, parsed_args)
+    clp_input_config = _generate_clp_io_config(logs_to_compress, parsed_args, clp_config)
     clp_output_config = OutputConfig.parse_obj(clp_config.archive_output)
     if parsed_args.tags:
         tag_list = [tag.strip().lower() for tag in parsed_args.tags.split(",") if tag]