diff --git a/components/clp-package-utils/clp_package_utils/general.py b/components/clp-package-utils/clp_package_utils/general.py
index d7800ea214..bb4be0e5b4 100644
--- a/components/clp-package-utils/clp_package_utils/general.py
+++ b/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()
 
diff --git a/components/clp-package-utils/clp_package_utils/scripts/compress.py b/components/clp-package-utils/clp_package_utils/scripts/compress.py
index 2829076e5b..f957cbef16 100755
--- a/components/clp-package-utils/clp_package_utils/scripts/compress.py
+++ b/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
diff --git a/components/clp-package-utils/clp_package_utils/scripts/native/compress.py b/components/clp-package-utils/clp_package_utils/scripts/native/compress.py
index b71907eb26..fc6a7df1d4 100755
--- a/components/clp-package-utils/clp_package_utils/scripts/native/compress.py
+++ b/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
 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
+    clp_config: CLPConfig, logs_to_compress: List[str], parsed_args: argparse.Namespace
 ) -> 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 key 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(clp_config, logs_to_compress, parsed_args)
     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]
diff --git a/components/clp-package-utils/clp_package_utils/scripts/start_clp.py b/components/clp-package-utils/clp_package_utils/scripts/start_clp.py
index 3e751c6f08..e3f8b76b1f 100755
--- a/components/clp-package-utils/clp_package_utils/scripts/start_clp.py
+++ b/components/clp-package-utils/clp_package_utils/scripts/start_clp.py
@@ -597,10 +597,11 @@ def generic_start_scheduler(
         "--mount", str(mounts.clp_home),
     ]
     # fmt: on
-    necessary_mounts = [
-        mounts.logs_dir,
-    ]
-    if COMPRESSION_SCHEDULER_COMPONENT_NAME == component_name:
+    necessary_mounts = [mounts.logs_dir]
+    if (
+        COMPRESSION_SCHEDULER_COMPONENT_NAME == component_name
+        and StorageType.FS == clp_config.logs_input.type
+    ):
         necessary_mounts.append(mounts.input_logs_dir)
     for mount in necessary_mounts:
         if mount:
@@ -741,8 +742,9 @@ def generic_start_worker(
         mounts.clp_home,
         mounts.data_dir,
         mounts.logs_dir,
-        mounts.input_logs_dir,
     ]
+    if StorageType.FS == clp_config.logs_input.type:
+        necessary_mounts.append(mounts.input_logs_dir)
     if worker_specific_mount:
         necessary_mounts.extend(worker_specific_mount)
 
diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py
index 1fbf5cbe63..e01994e573 100644
--- a/components/clp-py-utils/clp_py_utils/clp_config.py
+++ b/components/clp-py-utils/clp_py_utils/clp_config.py
@@ -381,9 +381,16 @@ def dump_to_primitive_dict(self):
 
 class S3Storage(BaseModel):
     type: Literal[StorageType.S3.value] = StorageType.S3.value
-    staging_directory: pathlib.Path
     s3_config: S3Config
 
+    def dump_to_primitive_dict(self):
+        d = self.dict()
+        return d
+
+
+class OutputS3Storage(S3Storage):
+    staging_directory: pathlib.Path
+
     @validator("staging_directory")
     def validate_staging_directory(cls, field):
         if "" == field:
@@ -394,11 +401,15 @@ def make_config_paths_absolute(self, clp_home: pathlib.Path):
         self.staging_directory = make_config_path_absolute(clp_home, self.staging_directory)
 
     def dump_to_primitive_dict(self):
-        d = self.dict()
+        d = super().dump_to_primitive_dict()
         d["staging_directory"] = str(d["staging_directory"])
         return d
 
 
+class InputFsStorage(FsStorage):
+    directory: pathlib.Path = pathlib.Path("/")
+
+
 class ArchiveFsStorage(FsStorage):
     directory: pathlib.Path = CLP_DEFAULT_DATA_DIRECTORY_PATH / "archives"
 
@@ -407,15 +418,17 @@ class StreamFsStorage(FsStorage):
     directory: pathlib.Path = CLP_DEFAULT_DATA_DIRECTORY_PATH / "streams"
 
 
-class ArchiveS3Storage(S3Storage):
+class ArchiveS3Storage(OutputS3Storage):
     staging_directory: pathlib.Path = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-archives"
 
 
-class StreamS3Storage(S3Storage):
+class StreamS3Storage(OutputS3Storage):
     staging_directory: pathlib.Path = CLP_DEFAULT_DATA_DIRECTORY_PATH / "staged-streams"
 
 
-def _get_directory_from_storage_config(storage_config: Union[FsStorage, S3Storage]) -> pathlib.Path:
+def _get_directory_from_storage_config(
+    storage_config: Union[FsStorage, OutputS3Storage],
+) -> pathlib.Path:
     storage_type = storage_config.type
     if StorageType.FS == storage_type:
         return storage_config.directory
@@ -426,7 +439,7 @@ def _get_directory_from_storage_config(storage_config: Union[FsStorage, S3Storag
 
 
 def _set_directory_for_storage_config(
-    storage_config: Union[FsStorage, S3Storage], directory
+    storage_config: Union[FsStorage, OutputS3Storage], directory
 ) -> None:
     storage_type = storage_config.type
     if StorageType.FS == storage_type:
@@ -548,7 +561,7 @@ def validate_port(cls, field):
 class CLPConfig(BaseModel):
     execution_container: Optional[str] = None
 
-    input_logs_directory: pathlib.Path = pathlib.Path("/")
+    logs_input: Union[InputFsStorage, S3Storage] = InputFsStorage()
 
     package: Package = Package()
     database: Database = Database()
@@ -572,7 +585,8 @@ class CLPConfig(BaseModel):
     _os_release_file_path: pathlib.Path = PrivateAttr(default=OS_RELEASE_FILE_PATH)
 
     def make_config_paths_absolute(self, clp_home: pathlib.Path):
-        self.input_logs_directory = make_config_path_absolute(clp_home, self.input_logs_directory)
+        if StorageType.FS == self.logs_input.type:
+            self.logs_input.make_config_paths_absolute(clp_home)
         self.credentials_file_path = make_config_path_absolute(clp_home, self.credentials_file_path)
         self.archive_output.storage.make_config_paths_absolute(clp_home)
         self.stream_output.storage.make_config_paths_absolute(clp_home)
@@ -580,14 +594,15 @@ def make_config_paths_absolute(self, clp_home: pathlib.Path):
         self.logs_directory = make_config_path_absolute(clp_home, self.logs_directory)
         self._os_release_file_path = make_config_path_absolute(clp_home, self._os_release_file_path)
 
-    def validate_input_logs_dir(self):
-        # NOTE: This can't be a pydantic validator since input_logs_dir might be a package-relative
-        # path that will only be resolved after pydantic validation
-        input_logs_dir = self.input_logs_directory
-        if not input_logs_dir.exists():
-            raise ValueError(f"input_logs_directory '{input_logs_dir}' doesn't exist.")
-        if not input_logs_dir.is_dir():
-            raise ValueError(f"input_logs_directory '{input_logs_dir}' is not a directory.")
+    def validate_logs_input_config(self):
+        if StorageType.FS == self.logs_input.type:
+            # NOTE: This can't be a pydantic validator since input_logs_dir might be a
+            # package-relative path that will only be resolved after pydantic validation
+            input_logs_dir = self.logs_input.directory
+            if not input_logs_dir.exists():
+                raise ValueError(f"logs_input.directory '{input_logs_dir}' doesn't exist.")
+            if not input_logs_dir.is_dir():
+                raise ValueError(f"logs_input.directory '{input_logs_dir}' is not a directory.")
 
     def validate_archive_output_config(self):
         if (
@@ -675,10 +690,10 @@ def load_redis_credentials_from_file(self):
 
     def dump_to_primitive_dict(self):
         d = self.dict()
+        d["logs_input"] = self.logs_input.dump_to_primitive_dict()
         d["archive_output"] = self.archive_output.dump_to_primitive_dict()
         d["stream_output"] = self.stream_output.dump_to_primitive_dict()
         # Turn paths into primitive strings
-        d["input_logs_directory"] = str(self.input_logs_directory)
         d["credentials_file_path"] = str(self.credentials_file_path)
         d["data_directory"] = str(self.data_directory)
         d["logs_directory"] = str(self.logs_directory)
diff --git a/components/clp-py-utils/clp_py_utils/s3_utils.py b/components/clp-py-utils/clp_py_utils/s3_utils.py
index 12d3755e48..0c893841cd 100644
--- a/components/clp-py-utils/clp_py_utils/s3_utils.py
+++ b/components/clp-py-utils/clp_py_utils/s3_utils.py
@@ -13,41 +13,6 @@
 AWS_ENDPOINT = "amazonaws.com"
 
 
-def parse_s3_url(s3_url: str) -> Tuple[str, str, str]:
-    """
-    Parses the region_code, bucket, and key_prefix from the given S3 URL.
-    :param s3_url: A host-style URL or path-style URL.
-    :return: A tuple of (region_code, bucket, key_prefix).
-    :raise: ValueError if `s3_url` is not a valid host-style URL or path-style URL.
-    """
-
-    host_style_url_regex = re.compile(
-        r"https://(?P<bucket_name>[a-z0-9.-]+)\.s3(\.(?P<region_code>[a-z0-9-]+))?"
-        r"\.(?P<endpoint>[a-z0-9.-]+)/(?P<key_prefix>[^?]+).*"
-    )
-    match = host_style_url_regex.match(s3_url)
-
-    if match is None:
-        path_style_url_regex = re.compile(
-            r"https://s3(\.(?P<region_code>[a-z0-9-]+))?\.(?P<endpoint>[a-z0-9.-]+)/"
-            r"(?P<bucket_name>[a-z0-9.-]+)/(?P<key_prefix>[^?]+).*"
-        )
-        match = path_style_url_regex.match(s3_url)
-
-    if match is None:
-        raise ValueError(f"Unsupported URL format: {s3_url}")
-
-    region_code = match.group("region_code")
-    bucket_name = match.group("bucket_name")
-    endpoint = match.group("endpoint")
-    key_prefix = match.group("key_prefix")
-
-    if AWS_ENDPOINT != endpoint:
-        raise ValueError(f"Unsupported endpoint: {endpoint}")
-
-    return region_code, bucket_name, key_prefix
-
-
 def generate_s3_virtual_hosted_style_url(
     region_code: str, bucket_name: str, object_key: str
 ) -> str:
diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py
index 7de797bb01..0a02aaaff7 100644
--- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py
+++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py
@@ -134,13 +134,13 @@ def _process_s3_input(
     and adds their metadata to paths_to_compress_buffer.
     :param s3_input_config:
     :param paths_to_compress_buffer:
-    :raises: RuntimeError if input URL doesn't resolve to any objects.
+    :raises: RuntimeError if input prefix doesn't resolve to any objects.
     :raises: Propagates `s3_get_object_metadata`'s exceptions.
     """
 
     object_metadata_list = s3_get_object_metadata(s3_input_config)
     if len(object_metadata_list) == 0:
-        raise RuntimeError("Input URL doesn't resolve to any object")
+        raise RuntimeError("Input prefix doesn't resolve to any object")
 
     for object_metadata in object_metadata_list:
         paths_to_compress_buffer.add_file(object_metadata)
diff --git a/components/package-template/src/etc/clp-config.yml b/components/package-template/src/etc/clp-config.yml
index 3e86199353..8e4f177312 100644
--- a/components/package-template/src/etc/clp-config.yml
+++ b/components/package-template/src/etc/clp-config.yml
@@ -1,8 +1,11 @@
-## A path containing any logs you which to compress. Must be reachable by all
+## Location (e.g., directory) containing any logs you wish to compress. Must be reachable by all
 ## workers.
-## - This path will be exposed inside the container, so symbolic links to files
-##   outside this path will be ignored.
-#input_logs_directory: "/"
+#logs_input:
+#  type: "fs"
+#
+#  # NOTE: This directory will be exposed inside the container, so symbolic links to files outside
+#  # this directory will be ignored.
+#  directory: "/"
 #
 ## File containing credentials for services
 #credentials_file_path: "etc/credentials.yml"
diff --git a/docs/src/user-guide/guides-using-object-storage/clp-config.md b/docs/src/user-guide/guides-using-object-storage/clp-config.md
index 02e3b93607..2b85739b6c 100644
--- a/docs/src/user-guide/guides-using-object-storage/clp-config.md
+++ b/docs/src/user-guide/guides-using-object-storage/clp-config.md
@@ -6,6 +6,31 @@ To use object storage with CLP, follow the steps below to configure each use cas
 If CLP is already running, shut it down, update its configuration, and then start it again.
 :::
 
+## Configuration for input logs
+
+To configure CLP to compress logs from S3, update the `logs_input` key in
+`<package>/etc/clp-config.yml` with the values in the code block below, replacing the fields in
+angle brackets (`<>`) with the appropriate values:
+
+```yaml
+logs_input:
+  type: "s3"
+  s3_config:
+    region_code: "<region-code>"
+    bucket: "<bucket-name>"
+    key_prefix: "<key-prefix>"
+    credentials:
+      access_key_id: "<aws-access-key-id>"
+      secret_access_key: "<aws-secret-access-key>"
+```
+* `s3_config` configures both the S3 bucket where logs are to be retrieved from and the credentials
+  for accessing it.
+  * `<region-code>` is the AWS region [code][aws-region-codes] for the bucket.
+  * `<bucket-name>` is the bucket's name.
+  * `<key-prefix>` is the prefix of all logs you wish to compress and should be the same as the
+    `<all-logs-prefix>` value from the [compression IAM policy][compression-iam-policy].
+  * `credentials` contains the CLP IAM user's credentials.
+
 ## Configuration for archive storage
 
 To configure CLP to store archives on S3, update the `archive_output.storage` key in
@@ -76,3 +101,4 @@ future release.
 :::
 
 [aws-region-codes]: https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Concepts.RegionsAndAvailabilityZones.html#Concepts.RegionsAndAvailabilityZones.Availability
+[compression-iam-policy]: ./object-storage-config.md#configuration-for-compression
diff --git a/docs/src/user-guide/guides-using-object-storage/clp-usage.md b/docs/src/user-guide/guides-using-object-storage/clp-usage.md
index 6fab2db443..963e659121 100644
--- a/docs/src/user-guide/guides-using-object-storage/clp-usage.md
+++ b/docs/src/user-guide/guides-using-object-storage/clp-usage.md
@@ -5,48 +5,33 @@ should be able to use CLP as described in the [quick start](../quick-start-overv
 
 ## Compressing logs from S3
 
-To compress logs from S3, use the `s3` subcommand as follows, replacing the fields in angle brackets
-(`<>`) with the appropriate values:
+To compress logs from S3, use the `sbin/compress.sh` script as follows, replacing the fields in
+angle brackets (`<>`) with the appropriate values:
 
 ```bash
 sbin/compress.sh \
-  s3 \
-  --aws-credentials-file <credentials-file> \
   --timestamp-key <timestamp-key> \
-  https://<bucket-name>.s3.<region-code>.amazonaws.com/<prefix>
+  <prefix>
 ```
 
-* `<credentials-file>` is the path to an AWS credentials file like the following:
-
-  ```ini
-  [default]
-  aws_access_key_id = <aws-access-key-id>
-  aws_secret_access_key = <aws-secret-access-key>
-  ```
-
-  * CLP expects the credentials to be in the `default` section.
-  * `<aws-access-key-id>` and `<aws-secret-access-key>` are the access key ID and secret access
-    key of the CLP IAM user.
-  * If you don't want to use a credentials file, you can specify the credentials on the command
-    line using the `--aws-access-key-id` and `--aws-secret-access-key` flags (note that this may
-    expose your credentials to other users running on the system).
-
-* `<timestamp-key>` is the field path of the kv-pair that contains the timestamp in each log event.
-* `<bucket-name>` is the name of the S3 bucket containing your logs.
-* `<region-code>` is the AWS region [code][aws-region-codes] for the S3 bucket containing your logs.
-* `<prefix>` is the prefix of all logs you wish to compress and must begin with the
-  `<all-logs-prefix>` value from the [compression IAM policy][compression-iam-policy].
+* `<prefix>` is the prefix of all logs you wish to compress and must be relative to
+  [logs-input.s3_config.key_prefix][logs-input-s3-config].
+  * E.g., if you want to compress the S3 object `/a/b/c.jsonl`, and
+    `logs-input.s3_config.key_prefix` is `/a/`, then you would replace `<prefix>` in the command
+    above with `b/c.jsonl`.
 
 :::{note}
-The `s3` subcommand only supports a single URL but will compress any logs that have the given
+Compressing from S3 only supports a single prefix but will compress any logs that have the given
 prefix.
 
-If you wish to compress a single log file, specify the entire path to the log file. However, if that
-log file's path is a prefix of another log file's path, then both log files will be compressed
+If you wish to compress a single log file, specify the entire path to the log file
+(relative to `logs-input.s3_config.key_prefix`). However, if that log file's path is a
+prefix of another log file's path, then both log files will be compressed
 (e.g., with two files "logs/syslog" and "logs/syslog.1", a prefix like "logs/syslog" will cause
 both logs to be compressed). This limitation will be addressed in a future release.
 :::
 
 [add-iam-policy]: https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html#embed-inline-policy-console
 [aws-region-codes]: https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Concepts.RegionsAndAvailabilityZones.html#Concepts.RegionsAndAvailabilityZones.Availability
-[compression-iam-policy]: ./object-storage-config.md#configuration-for-compression
\ No newline at end of file
+[compression-iam-policy]: ./object-storage-config.md#configuration-for-compression
+[logs-input-s3-config]: ./clp-config.md#configuration-for-input-logs
diff --git a/docs/src/user-guide/quick-start-compression/json.md b/docs/src/user-guide/quick-start-compression/json.md
index 6091762a89..7784a4f509 100644
--- a/docs/src/user-guide/quick-start-compression/json.md
+++ b/docs/src/user-guide/quick-start-compression/json.md
@@ -3,10 +3,9 @@
 To compress JSON logs, from inside the package directory, run:
 
 ```bash
-sbin/compress.sh fs --timestamp-key '<timestamp-key>' <path1> [<path2> ...]
+sbin/compress.sh --timestamp-key '<timestamp-key>' <path1> [<path2> ...]
 ```
 
-* `fs` is a subcommand for compressing logs from the filesystem.
 * `<timestamp-key>` is the field path of the kv-pair that contains the timestamp in each log event.
     * E.g., if your log events look like
       `{"timestamp": {"iso8601": "2024-01-01 00:01:02.345", ...}}`, you should enter
diff --git a/docs/src/user-guide/quick-start-compression/text.md b/docs/src/user-guide/quick-start-compression/text.md
index 18179a65b1..29e798b9d8 100644
--- a/docs/src/user-guide/quick-start-compression/text.md
+++ b/docs/src/user-guide/quick-start-compression/text.md
@@ -3,7 +3,7 @@
 To compress unstructured text logs, from inside the package directory, run:
 
 ```bash
-sbin/compress.sh fs <path1> [<path2> ...]
+sbin/compress.sh <path1> [<path2> ...]
 ```
 
 `<path...>` are paths to unstructured text log files or directories containing such files.