diff --git a/components/clp-package-utils/clp_package_utils/scripts/dataset_manager.py b/components/clp-package-utils/clp_package_utils/scripts/dataset_manager.py new file mode 100644 index 0000000000..4e0d4fcf7e --- /dev/null +++ b/components/clp-package-utils/clp_package_utils/scripts/dataset_manager.py @@ -0,0 +1,186 @@ +import argparse +import logging +import subprocess +import sys +from pathlib import Path +from typing import Final, List + +from clp_py_utils.clp_config import ( + ARCHIVE_MANAGER_ACTION_NAME, + StorageEngine, + StorageType, +) +from clp_py_utils.s3_utils import generate_container_auth_options + +from clp_package_utils.general import ( + CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH, + dump_container_config, + generate_container_config, + generate_container_name, + generate_container_start_cmd, + get_clp_home, + load_config_file, + validate_and_load_db_credentials_file, + validate_dataset_name, +) + +# Command/Argument Constants +LIST_COMMAND: Final[str] = "list" +DEL_COMMAND: Final[str] = "del" + +logger = logging.getLogger(__file__) + + +def main(argv: List[str]) -> int: + clp_home = get_clp_home() + default_config_file_path = clp_home / CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH + + # Top-level parser and options + args_parser = argparse.ArgumentParser(description="List or delete datasets.") + args_parser.add_argument( + "--config", + "-c", + default=str(default_config_file_path), + help="CLP package configuration file.", + ) + args_parser.add_argument( + "--verbose", + "-v", + action="store_true", + help="Enable debug logging.", + ) + + # Top-level commands + subparsers = args_parser.add_subparsers( + dest="subcommand", + required=True, + ) + subparsers.add_parser( + LIST_COMMAND, + help="List existing datasets.", + ) + + # Options for delete subcommand + del_parser = subparsers.add_parser( + DEL_COMMAND, + help="Delete datasets from the database and file storage.", + ) + del_parser.add_argument( + "datasets", + nargs="*", + help="Datasets to delete.", + ) + del_parser.add_argument( + "-a", + "--all", + dest="del_all", + action="store_true", + help="Delete all existing datasets.", + ) + + parsed_args = args_parser.parse_args(argv[1:]) + subcommand = parsed_args.subcommand + if parsed_args.verbose: + logger.setLevel(logging.DEBUG) + else: + logger.setLevel(logging.INFO) + + # Validate and load config file + try: + config_file_path = Path(parsed_args.config) + clp_config = load_config_file(config_file_path, default_config_file_path, clp_home) + clp_config.validate_logs_dir() + + # Validate and load necessary credentials + validate_and_load_db_credentials_file(clp_config, clp_home, False) + except: + logger.exception("Failed to load config.") + return -1 + + storage_engine = clp_config.package.storage_engine + if StorageEngine.CLP_S != storage_engine: + logger.error(f"Datasets aren't supported for storage engine: {storage_engine}.") + return -1 + + # Validate input depending on subcommands + if DEL_COMMAND == subcommand: + datasets = parsed_args.datasets + del_all_flag = parsed_args.del_all + if not del_all_flag and len(datasets) == 0: + args_parser.error("No datasets specified for deletion.") + if del_all_flag and len(datasets) != 0: + args_parser.error("The -a/--all flag cannot be used together with a dataset name.") + + try: + clp_db_connection_params = clp_config.database.get_clp_connection_params_and_type(True) + table_prefix = clp_db_connection_params["table_prefix"] + for dataset in datasets: + validate_dataset_name(table_prefix, dataset) + except: + logger.exception("Invalid dataset name.") + return -1 + + container_name = generate_container_name("dataset-manager") + + container_clp_config, mounts = generate_container_config(clp_config, clp_home) + generated_config_path_on_container, generated_config_path_on_host = dump_container_config( + container_clp_config, clp_config, container_name + ) + + necessary_mounts = [ + mounts.clp_home, + mounts.logs_dir, + ] + if clp_config.archive_output.storage.type == StorageType.FS: + necessary_mounts.append(mounts.archives_output_dir) + + aws_mount, aws_env_vars = generate_container_auth_options( + clp_config, ARCHIVE_MANAGER_ACTION_NAME + ) + if aws_mount: + necessary_mounts.append(mounts.aws_config_dir) + + container_start_cmd = generate_container_start_cmd( + container_name, necessary_mounts, clp_config.execution_container + ) + + if len(aws_env_vars) != 0: + for aws_env_var in aws_env_vars: + container_start_cmd.extend(["-e", aws_env_var]) + + # fmt: off + dataset_manager_cmd = [ + "python3", + "-m", "clp_package_utils.scripts.native.dataset_manager", + "--config", str(generated_config_path_on_container), + ] + # fmt: on + + dataset_manager_cmd.append(subcommand) + + # Add subcommand-specific arguments + if DEL_COMMAND == subcommand: + if parsed_args.del_all: + dataset_manager_cmd.append("--all") + else: + dataset_manager_cmd.extend(parsed_args.datasets) + elif LIST_COMMAND != subcommand: + logger.error(f"Unsupported subcommand: `{subcommand}`.") + return -1 + + cmd = container_start_cmd + dataset_manager_cmd + + proc = subprocess.run(cmd) + ret_code = proc.returncode + if 0 != ret_code: + logger.error("Dataset manager failed.") + logger.debug(f"Docker command failed: {' '.join(cmd)}") + + # Remove generated files + generated_config_path_on_host.unlink() + + return ret_code + + +if "__main__" == __name__: + sys.exit(main(sys.argv)) diff --git a/components/clp-package-utils/clp_package_utils/scripts/native/dataset_manager.py b/components/clp-package-utils/clp_package_utils/scripts/native/dataset_manager.py new file mode 100644 index 0000000000..29f3177bef --- /dev/null +++ b/components/clp-package-utils/clp_package_utils/scripts/native/dataset_manager.py @@ -0,0 +1,245 @@ +import argparse +import logging +import shutil +import sys +from contextlib import closing +from pathlib import Path +from typing import Dict, List + +from clp_py_utils.clp_config import ArchiveOutput, Database, S3Config, StorageType +from clp_py_utils.clp_metadata_db_utils import ( + delete_dataset_from_metadata_db, + get_datasets_table_name, +) +from clp_py_utils.s3_utils import s3_delete_by_key_prefix +from clp_py_utils.sql_adapter import SQL_Adapter + +from clp_package_utils.general import ( + CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH, + CLPConfig, + get_clp_home, + load_config_file, +) +from clp_package_utils.scripts.dataset_manager import ( + DEL_COMMAND, + LIST_COMMAND, +) + +logger: logging.Logger = logging.getLogger(__file__) + + +def _get_dataset_info( + db_config: Database, +) -> Dict[str, str]: + """ + :param db_config: + :return: A map of name -> archive_storage_directory for each dataset that exists. + """ + + sql_adapter = SQL_Adapter(db_config) + with closing(sql_adapter.create_connection(True)) as db_conn, closing( + db_conn.cursor(dictionary=True) + ) as db_cursor: + clp_db_connection_params = db_config.get_clp_connection_params_and_type(True) + table_prefix = clp_db_connection_params["table_prefix"] + db_cursor.execute( + f"SELECT name, archive_storage_directory FROM `{get_datasets_table_name(table_prefix)}`" + ) + rows = db_cursor.fetchall() + return {row["name"]: row["archive_storage_directory"] for row in rows} + + +def _handle_list_datasets(datasets: Dict[str, str]) -> int: + logger.info(f"Found {len(datasets)} datasets.") + for dataset in datasets.keys(): + logger.info(dataset) + return 0 + + +def _handle_del_datasets( + clp_config: CLPConfig, + parsed_args: argparse.Namespace, + existing_datasets_info: Dict[str, str], +): + if len(existing_datasets_info) == 0: + logger.warning("No datasets exist.") + return 0 + + datasets_to_delete: Dict[str, str] = {} + if parsed_args.del_all: + datasets_to_delete = existing_datasets_info + else: + datasets = parsed_args.datasets + for dataset in datasets: + if dataset not in existing_datasets_info: + logger.error(f"Dataset `{dataset}` doesn't exist. Aborting deletion.") + return -1 + + datasets_to_delete = {dataset: existing_datasets_info[dataset] for dataset in datasets} + + for dataset, dataset_archive_storage_dir in datasets_to_delete.items(): + if not _delete_dataset(clp_config, dataset, dataset_archive_storage_dir): + return -1 + + return 0 + + +def _delete_dataset(clp_config: CLPConfig, dataset: str, dataset_archive_storage_dir: str) -> bool: + try: + _try_deleting_archives(clp_config.archive_output, dataset_archive_storage_dir) + logger.info(f"Deleted archives of dataset `{dataset}`.") + except: + logger.exception(f"Failed to delete archives of dataset `{dataset}`.") + return False + + try: + _delete_dataset_from_database(clp_config.database, dataset) + logger.info(f"Deleted dataset `{dataset}` from the metadata database.") + except: + logger.exception(f"Failed to delete dataset `{dataset}` from the metadata database.") + return False + + return True + + +def _try_deleting_archives( + archive_output_config: ArchiveOutput, dataset_archive_storage_dir: str +) -> None: + archive_storage_config = archive_output_config.storage + storage_type = archive_storage_config.type + if StorageType.FS == storage_type: + _try_deleting_archives_from_fs(archive_output_config, dataset_archive_storage_dir) + elif StorageType.S3 == storage_type: + _try_deleting_archives_from_s3( + archive_storage_config.s3_config, dataset_archive_storage_dir + ) + else: + raise ValueError(f"Unsupported storage type: {storage_type}") + + +def _try_deleting_archives_from_fs( + archive_output_config: ArchiveOutput, dataset_archive_storage_dir: str +) -> None: + archives_dir = archive_output_config.get_directory() + dataset_archive_storage_path = Path(dataset_archive_storage_dir).resolve() + if not dataset_archive_storage_path.is_relative_to(archives_dir): + raise ValueError( + f"'{dataset_archive_storage_path}' is not within top-level archive storage directory" + f" '{archives_dir}'" + ) + + if not dataset_archive_storage_path.exists(): + logger.debug(f"'{dataset_archive_storage_path}' doesn't exist.") + return + + if not dataset_archive_storage_path.is_dir(): + logger.debug(f"'{dataset_archive_storage_path}' isn't a directory. Skipping deletion.") + return + + shutil.rmtree(dataset_archive_storage_path) + + +def _try_deleting_archives_from_s3(s3_config: S3Config, archive_storage_key_prefix: str) -> None: + # Add trailing '/' to avoid deleting other datasets with similar prefixes + if not archive_storage_key_prefix.endswith("/"): + archive_storage_key_prefix += "/" + + s3_delete_by_key_prefix( + s3_config.region_code, + s3_config.bucket, + archive_storage_key_prefix, + s3_config.aws_authentication, + ) + + +def _delete_dataset_from_database(database_config: Database, dataset: str) -> None: + sql_adapter = SQL_Adapter(database_config) + + with closing(sql_adapter.create_connection(True)) as db_conn, closing( + db_conn.cursor(dictionary=True) + ) as db_cursor: + clp_db_connection_params = database_config.get_clp_connection_params_and_type(True) + table_prefix = clp_db_connection_params["table_prefix"] + delete_dataset_from_metadata_db(db_cursor, table_prefix, dataset) + db_conn.commit() + + +def main(argv: List[str]) -> int: + clp_home = get_clp_home() + default_config_file_path = clp_home / CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH + + # Top-level parser and options + args_parser = argparse.ArgumentParser(description="List or delete datasets.") + args_parser.add_argument( + "--config", + "-c", + default=str(default_config_file_path), + help="CLP configuration file.", + ) + args_parser.add_argument( + "--verbose", + "-v", + action="store_true", + help="Enable debug logging.", + ) + + # Top-level commands + subparsers = args_parser.add_subparsers( + dest="subcommand", + required=True, + ) + subparsers.add_parser( + LIST_COMMAND, + help="List existing datasets.", + ) + + # Options for delete subcommand + del_parser = subparsers.add_parser( + DEL_COMMAND, + help="Delete datasets from the database and file storage.", + ) + del_parser.add_argument( + "datasets", + nargs="*", + help="Datasets to delete.", + ) + del_parser.add_argument( + "-a", + "--all", + dest="del_all", + action="store_true", + help="Delete all existing datasets.", + ) + + parsed_args = args_parser.parse_args(argv[1:]) + if parsed_args.verbose: + logger.setLevel(logging.DEBUG) + else: + logger.setLevel(logging.INFO) + + # Validate and load config file + config_file_path = Path(parsed_args.config) + try: + clp_config = load_config_file(config_file_path, default_config_file_path, clp_home) + clp_config.validate_logs_dir() + except: + logger.exception("Failed to load config.") + return -1 + + try: + existing_datasets_info = _get_dataset_info(clp_config.database) + except: + logger.exception("Failed to fetch datasets from the database.") + return -1 + + if LIST_COMMAND == parsed_args.subcommand: + return _handle_list_datasets(existing_datasets_info) + elif DEL_COMMAND == parsed_args.subcommand: + return _handle_del_datasets(clp_config, parsed_args, existing_datasets_info) + else: + logger.error(f"Unsupported subcommand: `{parsed_args.subcommand}`.") + return -1 + + +if "__main__" == __name__: + sys.exit(main(sys.argv)) 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 7e4fcb23dd..c10806ebd5 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -67,6 +67,9 @@ | STORAGE_MANAGEMENT_COMPONENTS, } +# Action names +ARCHIVE_MANAGER_ACTION_NAME = "archive_manager" + QUERY_JOBS_TABLE_NAME = "query_jobs" QUERY_TASKS_TABLE_NAME = "query_tasks" COMPRESSION_JOBS_TABLE_NAME = "compression_jobs" diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index e6dd3c9d00..217db17676 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -261,6 +261,37 @@ def delete_archives_from_metadata_db( ) +def delete_dataset_from_metadata_db(db_cursor, table_prefix: str, dataset: str) -> None: + """ + Deletes all tables associated with `dataset` from the metadata database. + + :param db_cursor: + :param table_prefix: + :param dataset: + """ + + # Drop tables in an order such that no foreign key constraint is violated. + tables_in_removal_order = [ + get_column_metadata_table_name(table_prefix, dataset), + get_files_table_name(table_prefix, dataset), + get_archive_tags_table_name(table_prefix, dataset), + get_tags_table_name(table_prefix, dataset), + get_archives_table_name(table_prefix, dataset), + ] + + for table in tables_in_removal_order: + db_cursor.execute(f"DROP TABLE IF EXISTS `{table}`") + + # Remove the dataset row from the datasets table + db_cursor.execute( + f""" + DELETE FROM `{get_datasets_table_name(table_prefix)}` + WHERE name = %s + """, + (dataset,), + ) + + def get_archive_tags_table_name(table_prefix: str, dataset: str | None) -> str: return _get_table_name(table_prefix, ARCHIVE_TAGS_TABLE_SUFFIX, dataset) 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 9e376a4e68..e903112f8b 100644 --- a/components/clp-py-utils/clp_py_utils/s3_utils.py +++ b/components/clp-py-utils/clp_py_utils/s3_utils.py @@ -1,13 +1,14 @@ import os import re from pathlib import Path -from typing import Dict, List, Optional, Set, Tuple, Union +from typing import Dict, Final, List, Optional, Set, Tuple, Union import boto3 from botocore.config import Config from job_orchestration.scheduler.job_config import S3InputConfig from clp_py_utils.clp_config import ( + ARCHIVE_MANAGER_ACTION_NAME, AwsAuthentication, AwsAuthType, CLPConfig, @@ -26,10 +27,12 @@ from clp_py_utils.core import FileMetadata # Constants -AWS_ENDPOINT = "amazonaws.com" -AWS_ENV_VAR_ACCESS_KEY_ID = "AWS_ACCESS_KEY_ID" -AWS_ENV_VAR_SECRET_ACCESS_KEY = "AWS_SECRET_ACCESS_KEY" -AWS_ENV_VAR_SESSION_TOKEN = "AWS_SESSION_TOKEN" +AWS_ENDPOINT: Final[str] = "amazonaws.com" +AWS_ENV_VAR_ACCESS_KEY_ID: Final[str] = "AWS_ACCESS_KEY_ID" +AWS_ENV_VAR_SECRET_ACCESS_KEY: Final[str] = "AWS_SECRET_ACCESS_KEY" +AWS_ENV_VAR_SESSION_TOKEN: Final[str] = "AWS_SESSION_TOKEN" + +S3_OBJECT_DELETION_BATCH_SIZE_MAX: Final[int] = 1000 def _get_session_credentials(aws_profile: Optional[str] = None) -> Optional[S3Credentials]: @@ -96,29 +99,31 @@ def get_credential_env_vars(auth: AwsAuthentication) -> Dict[str, str]: def generate_container_auth_options( - clp_config: CLPConfig, component_type: str + clp_config: CLPConfig, container_type: str ) -> Tuple[bool, List[str]]: """ Generates Docker container authentication options for AWS S3 access based on the given type. Handles authentication methods that require extra configuration (profile, env_vars). :param clp_config: CLPConfig containing storage configurations. - :param component_type: Type of calling container (compression, log_viewer, or query). + :param container_type: Type of the calling container. :return: Tuple of (whether aws config mount is needed, credential env_vars to set). :raises: ValueError if environment variables are not set correctly. """ output_storages_by_component_type: List[Union[S3Storage, FsStorage]] = [] input_storage_needed = False - if component_type in ( + if container_type in ( COMPRESSION_SCHEDULER_COMPONENT_NAME, COMPRESSION_WORKER_COMPONENT_NAME, ): output_storages_by_component_type = [clp_config.archive_output.storage] input_storage_needed = True - elif component_type in (WEBUI_COMPONENT_NAME,): + elif container_type in (ARCHIVE_MANAGER_ACTION_NAME,): + output_storages_by_component_type = [clp_config.archive_output.storage] + elif container_type in (WEBUI_COMPONENT_NAME,): output_storages_by_component_type = [clp_config.stream_output.storage] - elif component_type in ( + elif container_type in ( GARBAGE_COLLECTOR_COMPONENT_NAME, QUERY_SCHEDULER_COMPONENT_NAME, QUERY_WORKER_COMPONENT_NAME, @@ -128,7 +133,7 @@ def generate_container_auth_options( clp_config.stream_output.storage, ] else: - raise ValueError(f"Component type {component_type} is not valid.") + raise ValueError(f"Container type {container_type} is not valid.") config_mount = False add_env_vars = False @@ -172,28 +177,28 @@ def generate_container_auth_options( return (config_mount, credentials_env_vars) -def _create_s3_client(s3_config: S3Config, boto3_config: Optional[Config] = None) -> boto3.client: - auth = s3_config.aws_authentication +def _create_s3_client( + region_code: str, s3_auth: AwsAuthentication, boto3_config: Optional[Config] = None +) -> boto3.client: aws_session: Optional[boto3.Session] = None - - if AwsAuthType.profile == auth.type: + if AwsAuthType.profile == s3_auth.type: aws_session = boto3.Session( - profile_name=auth.profile, - region_name=s3_config.region_code, + profile_name=s3_auth.profile, + region_name=region_code, ) - elif AwsAuthType.credentials == auth.type: - credentials = auth.credentials + elif AwsAuthType.credentials == s3_auth.type: + credentials = s3_auth.credentials aws_session = boto3.Session( aws_access_key_id=credentials.access_key_id, aws_secret_access_key=credentials.secret_access_key, - region_name=s3_config.region_code, + region_name=region_code, aws_session_token=credentials.session_token, ) - elif AwsAuthType.env_vars == auth.type or AwsAuthType.ec2 == auth.type: + elif AwsAuthType.env_vars == s3_auth.type or AwsAuthType.ec2 == s3_auth.type: # Use default session which will use environment variables or instance role - aws_session = boto3.Session(region_name=s3_config.region_code) + aws_session = boto3.Session(region_name=region_code) else: - raise ValueError(f"Unsupported authentication type: {auth.type}") + raise ValueError(f"Unsupported authentication type: {s3_auth.type}") s3_client = aws_session.client("s3", config=boto3_config) return s3_client @@ -260,7 +265,7 @@ def s3_get_object_metadata(s3_input_config: S3InputConfig) -> List[FileMetadata] :raises: Propagates `boto3.paginator`'s exceptions. """ - s3_client = _create_s3_client(s3_input_config) + s3_client = _create_s3_client(s3_input_config.region_code, s3_input_config.aws_authentication) file_metadata_list: List[FileMetadata] = list() paginator = s3_client.get_paginator("list_objects_v2") @@ -304,7 +309,7 @@ def s3_put(s3_config: S3Config, src_file: Path, dest_path: str) -> None: ) boto3_config = Config(retries=dict(total_max_attempts=3, mode="adaptive")) - s3_client = _create_s3_client(s3_config, boto3_config) + s3_client = _create_s3_client(s3_config.region_code, s3_config.aws_authentication, boto3_config) with open(src_file, "rb") as file_data: s3_client.put_object( @@ -312,6 +317,44 @@ def s3_put(s3_config: S3Config, src_file: Path, dest_path: str) -> None: ) +def s3_delete_by_key_prefix( + region_code: str, bucket_name: str, key_prefix: str, s3_auth: AwsAuthentication +) -> None: + """ + Deletes all objects under the S3 path `bucket_name`/`key_prefix`. + + :param region_code: + :param bucket_name: + :param key_prefix: + :param s3_auth: + :raises: ValueError if any parameter is invalid. + :raises: Propagates `boto3.client.delete_objects`'s exceptions. + """ + + if not bool(region_code): + raise ValueError("Region code is not specified") + if not bool(bucket_name): + raise ValueError("Bucket name is not specified") + if not bool(key_prefix): + raise ValueError("Key prefix is not specified") + + boto3_config = Config(retries=dict(total_max_attempts=3, mode="adaptive")) + s3_client = _create_s3_client(region_code, s3_auth, boto3_config) + + paginator = s3_client.get_paginator("list_objects_v2") + for page in paginator.paginate( + Bucket=bucket_name, + Prefix=key_prefix, + PaginationConfig={"PageSize": S3_OBJECT_DELETION_BATCH_SIZE_MAX}, + ): + contents = page.get("Contents", None) + if contents is None: + continue + + deletion_config = {"Objects": [{"Key": obj["Key"]} for obj in contents]} + s3_client.delete_objects(Bucket=bucket_name, Delete=deletion_config) + + def s3_delete_objects(s3_config: S3Config, object_keys: Set[str]) -> None: """ Deletes objects from an S3 bucket. The objects are identified by keys relative to @@ -327,8 +370,7 @@ def s3_delete_objects(s3_config: S3Config, object_keys: Set[str]) -> None: :raises: Propagates `boto3.client.delete_object`'s exceptions. """ boto3_config = Config(retries=dict(total_max_attempts=3, mode="adaptive")) - s3_client = _create_s3_client(s3_config, boto3_config) - max_num_objects_to_delete_per_batch = 1000 + s3_client = _create_s3_client(s3_config.region_code, s3_config.aws_authentication, boto3_config) def _gen_deletion_config(objects_list: List[str]): return {"Objects": [{"Key": object_to_delete} for object_to_delete in objects_list]} @@ -336,7 +378,7 @@ def _gen_deletion_config(objects_list: List[str]): objects_to_delete: List[str] = [] for relative_obj_key in object_keys: objects_to_delete.append(s3_config.key_prefix + relative_obj_key) - if len(objects_to_delete) < max_num_objects_to_delete_per_batch: + if len(objects_to_delete) < S3_OBJECT_DELETION_BATCH_SIZE_MAX: continue s3_client.delete_objects( 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 ec2018a2d6..3f0e6731b5 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -158,7 +158,6 @@ def search_and_schedule_new_tasks( db_cursor, clp_metadata_db_connection_config: Dict[str, Any], clp_archive_output: ArchiveOutput, - existing_datasets: Set[str], ): """ For all jobs with PENDING status, splits the job into tasks and schedules them. @@ -166,10 +165,13 @@ def search_and_schedule_new_tasks( :param db_cursor: :param clp_metadata_db_connection_config: :param clp_archive_output: - :param existing_datasets: """ global scheduled_jobs + existing_datasets = fetch_existing_datasets( + db_cursor, clp_metadata_db_connection_config["table_prefix"] + ) + logger.debug("Search and schedule new tasks") # Poll for new compression jobs @@ -194,7 +196,7 @@ def search_and_schedule_new_tasks( clp_archive_output, ) - # NOTE: This assumes we never delete a dataset + # NOTE: This assumes we never delete a dataset when compression jobs are being scheduled existing_datasets.add(dataset) paths_to_compress_buffer = PathsToCompressBuffer( @@ -421,11 +423,6 @@ def main(argv): clp_metadata_db_connection_config = ( sql_adapter.database_config.get_clp_connection_params_and_type(True) ) - existing_datasets: Set[str] = set() - if StorageEngine.CLP_S == clp_config.package.storage_engine: - existing_datasets = fetch_existing_datasets( - db_cursor, clp_metadata_db_connection_config["table_prefix"] - ) # Start Job Processing Loop while True: @@ -435,7 +432,6 @@ def main(argv): db_cursor, clp_metadata_db_connection_config, clp_config.archive_output, - existing_datasets, ) poll_running_jobs(db_conn, db_cursor) time.sleep(clp_config.compression_scheduler.jobs_poll_delay) diff --git a/components/package-template/src/sbin/admin-tools/dataset-manager.sh b/components/package-template/src/sbin/admin-tools/dataset-manager.sh new file mode 100755 index 0000000000..e0ecb767a1 --- /dev/null +++ b/components/package-template/src/sbin/admin-tools/dataset-manager.sh @@ -0,0 +1,9 @@ +#!/usr/bin/env bash + +script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )" +package_root="$script_dir/../.." + +PYTHONPATH=$(readlink -f "$package_root/lib/python3/site-packages") \ + python3 \ + -m clp_package_utils.scripts.dataset_manager \ + "$@" diff --git a/docs/src/user-guide/guides-using-object-storage/object-storage-config.md b/docs/src/user-guide/guides-using-object-storage/object-storage-config.md index b6110e9fdc..8a37537eea 100644 --- a/docs/src/user-guide/guides-using-object-storage/object-storage-config.md +++ b/docs/src/user-guide/guides-using-object-storage/object-storage-config.md @@ -67,6 +67,18 @@ the fields in angle brackets (`<>`) with the appropriate values: "Resource": [ "arn:aws:s3::://*" ] + }, + { + "Effect": "Allow", + "Action": "s3:ListBucket", + "Resource": [ + "arn:aws:s3:::" + ], + "Condition": { + "StringLike": { + "s3:prefix": "/*" + } + } } ] }