From 4a5cde9ee4daeaa55c69150dbf64495b4a5f29f6 Mon Sep 17 00:00:00 2001 From: qiagu Date: Mon, 28 Feb 2022 10:59:08 -0800 Subject: [PATCH 01/43] Add boto3 as a runner dependency --- lib/galaxy/dependencies/__init__.py | 3 +++ lib/galaxy/dependencies/conditional-requirements.txt | 3 +++ 2 files changed, 6 insertions(+) diff --git a/lib/galaxy/dependencies/__init__.py b/lib/galaxy/dependencies/__init__.py index 938aecb1a8d..67803280216 100644 --- a/lib/galaxy/dependencies/__init__.py +++ b/lib/galaxy/dependencies/__init__.py @@ -205,6 +205,9 @@ class ConditionalDependencies: def check_chronos_python(self): return "galaxy.jobs.runners.chronos:ChronosJobRunner" in self.job_runners + def check_boto3_python(self): + return "galaxy.jobs.runners.aws:AWSBatchJobRunner" in self.job_runners + def check_fluent_logger(self): return asbool(self.config["fluent_log"]) diff --git a/lib/galaxy/dependencies/conditional-requirements.txt b/lib/galaxy/dependencies/conditional-requirements.txt index 9909b5aeb45..203badfcaf5 100644 --- a/lib/galaxy/dependencies/conditional-requirements.txt +++ b/lib/galaxy/dependencies/conditional-requirements.txt @@ -54,3 +54,6 @@ tensorflow==2.7.2 # Run run.sh or common_startup script with GALAXY_DEPENDENCIES_INSTALL_WEASYPRINT=1 # to install weasyprint as part of Galaxy's conditonal dependency instalation process. weasyprint + +# AWS Batch runner +boto3 From a9b17fa9d7e54f7e9bcfee91f0ea49a43018ba47 Mon Sep 17 00:00:00 2001 From: qiagu Date: Mon, 28 Feb 2022 23:50:52 -0800 Subject: [PATCH 02/43] Add aws_batch job runner --- lib/galaxy/jobs/runners/aws.py | 423 +++++++++++++++++++++++++++++++++ 1 file changed, 423 insertions(+) create mode 100644 lib/galaxy/jobs/runners/aws.py diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py new file mode 100644 index 00000000000..5fbbe3d80ae --- /dev/null +++ b/lib/galaxy/jobs/runners/aws.py @@ -0,0 +1,423 @@ +import functools +import hashlib +import logging +import os +import time + +from queue import Empty +from galaxy import model +from galaxy.jobs.runners import ( + AsynchronousJobRunner, + AsynchronousJobState, +) +from galaxy.util import smart_str, unicodify + +BOTO3_IMPORT_MSG = ( + "The Python 'boto3' package is required to use " + "this feature, please install it or correct the " + "following error:\nImportError {msg!s}" +) + +try: + import boto3 + +except ImportError as e: + boto3 = None + BOTO3_IMPORT_MSG.format(msg=unicodify(e)) + + +__all__ = ("AWSBatchJobRunner",) +LOGGER = logging.getLogger(__name__) + +STOP_SIGNAL = object() + + +class AWSBatchRunnerException(Exception): + pass + + +def handle_exception_call(func): + # Catch boto3 exceptions. + + @functools.wraps(func) + def wrapper(*args, **kwargs): + try: + return func(*args, **kwargs) + except Exception as e: + LOGGER.error(unicodify(e)) + + return wrapper + + +def to_dict(segments, v): + if len(segments) == 0: + return v + return {segments[0]: to_dict(segments[1:], v)} + + +def _write_logfile(logfile, msg): + with open(logfile, "w") as fil: + fil.write(msg) + + +def _add_galaxy_environment_variables(cpus, memory): + # Set: + # GALAXY_SLOTS: to docker_cpu + # GALAXY_MEMORY_MB to docker_memory + return [{"name": "GALAXY_SLOTS", "value": cpus}, {"name": "GALAXY_MEMORY_MB", "value": memory}] + + +class AWSBatchJobRunner(AsynchronousJobRunner): + runner_name = "AWSBatchRunner" + RUNNER_PARAM_SPEC_KEY = "runner_param_specs" + JOB_NAME_PREFIX = "galaxy-" + # AWS Batch queries up to 100 jobs at once. + MAX_JOBS_PER_QUERY = 100 + + RUNNER_PARAM_SPEC = { + "aws_access_key_id": { + "map": str, + }, + "aws_secret_access_key": { + "map": str, + } + } + + DESTINATION_PARAMS_SPEC = { + "vcpu": { + "default": 1.0, + "map_name": "vcpu", + "map": str, + }, + "memory": { + "default": 2048, + "map_name": "memory", + "map": str, + }, + "job_queue": { + "default": None, + "map_name": "job_queue", + "map": str, + }, + "job_role_arn": { + "default": None, + "map_name": "job_role_arn", + "map": str, + }, + "efs_filesystem_id": { + "default": None, + "map_name": "efs_filesystem_id", + "map": str, + }, + "efs_mount_point": { + "default": None, + "map_name": "efs_mount_point", + "map": str, + }, + "fargate_version": { + "default": None, + "map_name": "fargate_version", + "map": str, + } + } + + def __init__(self, app, nworkers, **kwargs): + """Initialize this job runner and start the monitor thread""" + assert boto3, BOTO3_IMPORT_MSG + if self.RUNNER_PARAM_SPEC_KEY not in kwargs: + kwargs[self.RUNNER_PARAM_SPEC_KEY] = {} + kwargs[self.RUNNER_PARAM_SPEC_KEY].update(self.RUNNER_PARAM_SPEC) + super().__init__(app, nworkers, **kwargs) + session = boto3.Session( + aws_access_key_id=self.runner_params.get('aws_access_key_id') or None, + aws_secret_access_key=self.runner_params.get('aws_secret_access_key') or None + ) + self._batch_client = session.client('batch') + + @handle_exception_call + def queue_job(self, job_wrapper): + LOGGER.debug(f"Starting queue_job for job {job_wrapper.get_id_tag()}") + if not self.prepare_job(job_wrapper, include_metadata=False, modify_command_for_container=False): + LOGGER.debug(f"Not ready {job_wrapper.get_id_tag()}") + return + + job_destination = job_wrapper.job_destination + if not job_destination.params.get("docker_enabled"): + raise AWSBatchRunnerException("AWSBatchJobRunner needs 'docker_enabled' to be set as True") + + destination_params = self.parse_destination_params(job_destination.params) + job_def = self._get_job_definition(job_wrapper, destination_params) + if 'fargate_version' in destination_params: + job_name, job_id = self._submit_fargate_job(job_def, job_wrapper, destination_params) + else: + job_name, job_id = self._submit_ec2_job(job_def, job_wrapper, destination_params) + + ajs = AsynchronousJobState( + files_dir=job_wrapper.working_directory, + job_wrapper=job_wrapper, + job_name=job_name, + job_id=job_id, + job_destination=job_destination, + ) + self.monitor_queue.put(ajs) + + def _get_job_definition(self, job_wrapper, destination_params): + tool_id = job_wrapper.tool.id + tool_version = job_wrapper.tool.version + h = hashlib.new("sha256") + h.update(smart_str(tool_version)) + compute_type = 'fargate' if 'fargate_version' in destination_params else 'ec2' + + jd_name = f"galaxy_tool__{tool_id}__{h.hexdigest()}__{compute_type}" + res = self._batch_client.describe_job_definitions( + jobDefinitionName=jd_name, + status="ACTIVE" + ) + if not res['jobDefinitions']: + docker_image = docker_image = self._find_container(job_wrapper).container_id + # user_id = job_wrapper.user.id + jd_arn = self._register_job_definition(jd_name, docker_image, destination_params) + else: + jd_arn = res['jobDefinitions'][0]['jobDefinitionArn'] + + return jd_arn + + def _register_job_definition(self, jd_name, docker_image, destination_params): + if 'fargate_version' in destination_params: + return self._register_fargate_job_definition(jd_name, docker_image, destination_params) + else: + return self._register_ec2_job_definition(jd_name, docker_image, destination_params) + + def _register_fargate_job_definition(self, jd_name, docker_image, destination_params): + res = self._batch_client.register_job_definition( + jobDefinitionName=jd_name, + type='container', + platformCapabilities=['FARGATE'], + containerProperties={ + 'image': docker_image, + 'command': [ + '/bin/sh', + ], + 'jobRoleArn': destination_params.get('job_role_arn'), + 'executionRoleArn': destination_params.get('execute_role_arn', None) or destination_params.get('job_role_arn'), + 'volumes': [ + { + 'name': 'efs_whole', + 'efsVolumeConfiguration': { + 'fileSystemId': destination_params.get('efs_filesystem_id'), + 'rootDirectory': '/', + 'transitEncryption': 'ENABLED', + 'authorizationConfig': { + 'iam': 'ENABLED' + } + } + }, + ], + 'mountPoints': [ + { + 'containerPath': destination_params.get('efs_mount_point'), + 'readOnly': False, + 'sourceVolume': 'efs_whole' + }, + ], + 'resourceRequirements': [ + { + 'type': 'VCPU', + 'value': destination_params.get('vcpu') + }, + { + 'type': 'MEMORY', + 'value': destination_params.get('memory') + } + ], + 'fargatePlatformConfiguration': { + 'platformVersion': destination_params.get('fargate_version') + } + } + ) + + assert res['ResponseMetadata']['HTTPStatusCode'] == 200 + + return res['jobDefinitionArn'] + + + def _register_ec2_job_definition(self, jd_name, docker_image, user_id, destination_params): + return '' + + def _submit_fargate_job(self, job_def, job_wrapper, destination_params): + job_name = self.JOB_NAME_PREFIX + job_wrapper.get_id_tag() + command_script_path = self.write_command(job_wrapper) + + res = self._batch_client.submit_job( + jobName=job_name, + jobQueue=destination_params.get('job_queue'), + jobDefinition=job_def, + containerOverrides={ + 'command': [ + f'$SHELL {command_script_path}', + ] + } + ) + + assert res['ResponseMetadata']['HTTPStatusCode'] == 200 + + return job_name, res['jobId'] + + def _submit_ec2_job(self, job_def, job_wrapper, destination_params): + pass + + @handle_exception_call + def stop_job(self, job_state): + job_id = job_state.job_id + + res = self._batch_client.describe_jobs + self._batch_client.terminate_job(jobId=job_id, reason="Killed by Galaxy!") + msg = "Job {name!r} is terminated" + LOGGER.debug(msg.format(name=job_state.job_name)) + + def recover(self, job, job_wrapper): + msg = "(name!r/runner!r) is still in {state!s} state, adding to" " the runner monitor queue" + job_id = job.get_job_runner_external_id() + ajs = AsynchronousJobState(files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper) + ajs.job_id = self.JOB_NAME_PREFIX + str(job_id) + ajs.command_line = job.command_line + ajs.job_wrapper = job_wrapper + ajs.job_destination = job_wrapper.job_destination + if job.state in (model.Job.states.RUNNING, model.Job.states.STOPPED): + LOGGER.debug(msg.format(name=job.id, runner=job.job_runner_external_id, state=job.state)) + ajs.old_state = model.Job.states.RUNNING + ajs.running = True + self.monitor_queue.put(ajs) + elif job.state == model.Job.states.QUEUED: + LOGGER.debug(msg.format(name=job.id, runner=job.job_runner_external_id, state="queued")) + ajs.old_state = model.Job.states.QUEUED + ajs.running = False + self.monitor_queue.put(ajs) + + def fail_job(self, job_state, exception=False): + if getattr(job_state, "stop_job", True): + self.stop_job(job_state) + job_state.job_wrapper.reclaim_ownership() + self._handle_runner_state("failure", job_state) + if not job_state.runner_state_handled: + job_state.job_wrapper.fail(getattr(job_state, "fail_message", "Job failed"), exception=exception) + self._finish_or_resubmit_job(job_state, "", job_state.fail_message, job_id=job_state.job_id) + if job_state.job_wrapper.cleanup_job == "always": + job_state.cleanup() + + def monitor(self): + """ + Watches jobs currently in the monitor queue and deals with state + changes (queued to running) and job completion. + """ + while True: + # Take any new watched jobs and put them on the monitor list + try: + while True: + async_job_state = self.monitor_queue.get_nowait() + if async_job_state is STOP_SIGNAL: + # TODO: This is where any cleanup would occur + self.handle_stop() + return + self.watched.append((async_job_state.job_id, async_job_state)) + except Empty: + pass + # Iterate over the list of watched jobs and check state + try: + self.check_watched_items() + except Exception: + LOGGER.exception("Unhandled exception checking active jobs") + # Sleep a bit before the next state check + time.sleep(self.app.config.job_runner_monitor_sleep) + + @handle_exception_call + def check_watched_items(self): + self.check_watched_items_by_batch(0, len(self.watched)) + + def check_watched_items_by_batch(self, start: int, end: int): + jobs = self.watched[start: self.MAX_JOBS_PER_QUERY] + if not jobs: + return + + jobs_dict = dict(jobs) + + res = self._batch_client.describe_jobs(jobs=list(jobs_dict.keys())) + + gotten = [] + for job in res['jobs']: + status = job['status'] + job_id = job['jobId'] + gotten.append(job_id) + job_state = jobs_dict[job_id] + + if status == 'SUCCEEDED': + self.watched.remove((job_id, job_state)) + start -= 1 + self._mark_as_successful(job_state) + elif status == 'FAILED': + self.watched.remove((job_id, job_state)) + start -= 1 + reason = job['statusReason'] + self._mark_as_failed(job_state, reason) + elif status in ('SUBMITTED', 'PENDING', 'RUNNABLE', 'STARTING', 'RUNNING'): + self._mark_as_active(job_state) + # TODO else? + + for job_id in set(jobs_dict.keys()) - set(gotten): + job_state = jobs_dict[job_id] + self.watched.remove((job_id, job_state)) + start -= 1 + reason = f"The track of Job {job_state} was lost for unknown reason!" + self._mark_as_failed(job_state, reason) + + self.check_watched_items_by_batch(start+self.MAX_JOBS_PER_QUERY, end) + + def _mark_as_successful(self, job_state): + msg = "Job {name!r} finished successfully" + _write_logfile(job_state.output_file, msg.format(name=job_state.job_name)) + _write_logfile(job_state.error_file, "") + job_state.running = False + job_state.job_wrapper.change_state(model.Job.states.OK) + self.mark_as_finished(job_state) + + def _mark_as_active(self, job_state): + job_state.running = True + job_state.job_wrapper.change_state(model.Job.states.RUNNING) + + def _mark_as_failed(self, job_state, reason): + _write_logfile(job_state.error_file, reason) + job_state.running = False + job_state.stop_job = True + job_state.job_wrapper.change_state(model.Job.states.ERROR) + job_state.fail_message = reason + self.mark_as_failed(job_state) + + # @handle_exception_call + # def finish_job(self, job_state): + # super().finish_job(job_state) + # self._batch_client.delete(job_state.job_id) + + def parse_destination_params(self, params): + parsed_params = {} + for k, spec in self.DESTINATION_PARAMS_SPEC.items(): + value = params.get(k, spec.get("default")) + map_to = spec.get("map_name") + mapper = spec.get("map") + segments = map_to.split("/") + parsed_params.update(to_dict(segments, mapper(value))) + return parsed_params + + def write_command(self, job_wrapper): + # Create command script instead passing it in the container + # preventing wrong characters parsing. + if not os.path.exists(job_wrapper.working_directory): + LOGGER.error("No working directory found") + + path = f"{job_wrapper.working_directory}/galaxy_{job_wrapper.get_id_tag()}.sh" + mode = 0o755 + + with open(path, "w", encoding="utf-8") as f: + f.write("#!/bin/bash\n") + f.write(job_wrapper.runner_command_line) + os.chmod(path, mode) + return path From e8ae98bc6646b187e012104a4052c6ea68793534 Mon Sep 17 00:00:00 2001 From: qiagu Date: Tue, 1 Mar 2022 11:00:43 -0800 Subject: [PATCH 03/43] Set minimum query interval --- lib/galaxy/jobs/runners/aws.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 5fbbe3d80ae..0d71e2dc53f 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -73,6 +73,8 @@ class AWSBatchJobRunner(AsynchronousJobRunner): JOB_NAME_PREFIX = "galaxy-" # AWS Batch queries up to 100 jobs at once. MAX_JOBS_PER_QUERY = 100 + # Higher minimum interval as jobs are queried in batches. + MIN_QUERY_INTERVAL = 10 RUNNER_PARAM_SPEC = { "aws_access_key_id": { @@ -328,7 +330,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): except Exception: LOGGER.exception("Unhandled exception checking active jobs") # Sleep a bit before the next state check - time.sleep(self.app.config.job_runner_monitor_sleep) + time.sleep(max(self.app.config.job_runner_monitor_sleep, self.MIN_QUERY_INTERVAL)) @handle_exception_call def check_watched_items(self): From b327e71454cbf252ee2765909ba58fd3228a53a7 Mon Sep 17 00:00:00 2001 From: qiagu Date: Wed, 2 Mar 2022 16:12:52 -0800 Subject: [PATCH 04/43] Enable container public IP --- lib/galaxy/jobs/runners/aws.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 0d71e2dc53f..0291f4f7d89 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -232,6 +232,12 @@ class AWSBatchJobRunner(AsynchronousJobRunner): 'value': destination_params.get('memory') } ], + 'logConfiguration': { + 'logDriver': 'awslogs' + }, + 'networkConfiguration': { + 'assignPublicIp': 'ENABLED' + }, 'fargatePlatformConfiguration': { 'platformVersion': destination_params.get('fargate_version') } @@ -256,7 +262,8 @@ class AWSBatchJobRunner(AsynchronousJobRunner): jobDefinition=job_def, containerOverrides={ 'command': [ - f'$SHELL {command_script_path}', + '/bin/bash', + f'{command_script_path}', ] } ) @@ -337,7 +344,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): self.check_watched_items_by_batch(0, len(self.watched)) def check_watched_items_by_batch(self, start: int, end: int): - jobs = self.watched[start: self.MAX_JOBS_PER_QUERY] + jobs = self.watched[start: start+self.MAX_JOBS_PER_QUERY] if not jobs: return From eef6a910ca2b6ff76044c9a412fe1ffe8d169629 Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 3 Mar 2022 12:29:55 -0800 Subject: [PATCH 05/43] Add environment variables and fix stop_job --- lib/galaxy/jobs/runners/aws.py | 46 +++++++++++++++++++--------------- 1 file changed, 26 insertions(+), 20 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 0291f4f7d89..8ced56a5879 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -62,9 +62,10 @@ def _write_logfile(logfile, msg): def _add_galaxy_environment_variables(cpus, memory): # Set: - # GALAXY_SLOTS: to docker_cpu - # GALAXY_MEMORY_MB to docker_memory - return [{"name": "GALAXY_SLOTS", "value": cpus}, {"name": "GALAXY_MEMORY_MB", "value": memory}] + # GALAXY_SLOTS, round 0.25 vpuc to 1. + # GALAXY_MEMORY_MB + return [{"name": "GALAXY_SLOTS", "value": str(int(max(cpus, 1)))}, + {"name": "GALAXY_MEMORY_MB", "value": str(memory)}] class AWSBatchJobRunner(AsynchronousJobRunner): @@ -89,12 +90,12 @@ class AWSBatchJobRunner(AsynchronousJobRunner): "vcpu": { "default": 1.0, "map_name": "vcpu", - "map": str, + "map": float, }, "memory": { "default": 2048, "map_name": "memory", - "map": str, + "map": int, }, "job_queue": { "default": None, @@ -154,6 +155,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): else: job_name, job_id = self._submit_ec2_job(job_def, job_wrapper, destination_params) + job_wrapper.set_external_id(job_id) ajs = AsynchronousJobState( files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper, @@ -225,13 +227,17 @@ class AWSBatchJobRunner(AsynchronousJobRunner): 'resourceRequirements': [ { 'type': 'VCPU', - 'value': destination_params.get('vcpu') + 'value': str(destination_params.get('vcpu')) }, { 'type': 'MEMORY', - 'value': destination_params.get('memory') + 'value': str(destination_params.get('memory')) } ], + 'environment': _add_galaxy_environment_variables( + destination_params.get('vcpu'), + destination_params.get('memory') + ), 'logConfiguration': { 'logDriver': 'awslogs' }, @@ -245,10 +251,8 @@ class AWSBatchJobRunner(AsynchronousJobRunner): ) assert res['ResponseMetadata']['HTTPStatusCode'] == 200 - return res['jobDefinitionArn'] - def _register_ec2_job_definition(self, jd_name, docker_image, user_id, destination_params): return '' @@ -269,43 +273,45 @@ class AWSBatchJobRunner(AsynchronousJobRunner): ) assert res['ResponseMetadata']['HTTPStatusCode'] == 200 - return job_name, res['jobId'] def _submit_ec2_job(self, job_def, job_wrapper, destination_params): pass @handle_exception_call - def stop_job(self, job_state): - job_id = job_state.job_id + def stop_job(self, job_wrapper): + job = job_wrapper.get_job() + external_id = job.get_job_runner_external_id() + job_name = self.JOB_NAME_PREFIX + job_wrapper.get_id_tag() - res = self._batch_client.describe_jobs - self._batch_client.terminate_job(jobId=job_id, reason="Killed by Galaxy!") + self._batch_client.terminate_job(jobId=external_id, reason="Terminated by Galaxy!") msg = "Job {name!r} is terminated" - LOGGER.debug(msg.format(name=job_state.job_name)) + LOGGER.debug(msg.format(name=job_name)) def recover(self, job, job_wrapper): msg = "(name!r/runner!r) is still in {state!s} state, adding to" " the runner monitor queue" job_id = job.get_job_runner_external_id() + job_name = self.JOB_NAME_PREFIX + job_wrapper.get_id_tag() ajs = AsynchronousJobState(files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper) - ajs.job_id = self.JOB_NAME_PREFIX + str(job_id) + ajs.job_id = str(job_id) + ajs.job_name = job_name ajs.command_line = job.command_line ajs.job_wrapper = job_wrapper ajs.job_destination = job_wrapper.job_destination if job.state in (model.Job.states.RUNNING, model.Job.states.STOPPED): - LOGGER.debug(msg.format(name=job.id, runner=job.job_runner_external_id, state=job.state)) + LOGGER.debug(msg.format(name=job.id, runner=job.job_runner_name, state=job.state)) ajs.old_state = model.Job.states.RUNNING ajs.running = True self.monitor_queue.put(ajs) elif job.state == model.Job.states.QUEUED: - LOGGER.debug(msg.format(name=job.id, runner=job.job_runner_external_id, state="queued")) + LOGGER.debug(msg.format(name=job.id, runner=job.job_runner_name, state="queued")) ajs.old_state = model.Job.states.QUEUED ajs.running = False self.monitor_queue.put(ajs) def fail_job(self, job_state, exception=False): if getattr(job_state, "stop_job", True): - self.stop_job(job_state) + self.stop_job(job_state.job_wrapper) job_state.job_wrapper.reclaim_ownership() self._handle_runner_state("failure", job_state) if not job_state.runner_state_handled: @@ -396,7 +402,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): def _mark_as_failed(self, job_state, reason): _write_logfile(job_state.error_file, reason) job_state.running = False - job_state.stop_job = True + job_state.stop_job = False job_state.job_wrapper.change_state(model.Job.states.ERROR) job_state.fail_message = reason self.mark_as_failed(job_state) From 4d96db977dd76d89ea2056d00c5c49cd0e8a57f0 Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 3 Mar 2022 18:43:27 -0800 Subject: [PATCH 06/43] Change job definition hash --- lib/galaxy/jobs/runners/aws.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 8ced56a5879..fa2445b9950 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -169,16 +169,19 @@ class AWSBatchJobRunner(AsynchronousJobRunner): tool_id = job_wrapper.tool.id tool_version = job_wrapper.tool.version h = hashlib.new("sha256") + h.update(smart_str(tool_id)) h.update(smart_str(tool_version)) - compute_type = 'fargate' if 'fargate_version' in destination_params else 'ec2' + for k, v in destination_params.items(): + h.update(smart_str(k+str(v))) + queue_name = destination_params.get('job_queue').rsplit('/', 1)[-1] - jd_name = f"galaxy_tool__{tool_id}__{h.hexdigest()}__{compute_type}" + jd_name = f"galaxy_tool__{tool_id}__{h.hexdigest()}__{queue_name}" res = self._batch_client.describe_job_definitions( jobDefinitionName=jd_name, status="ACTIVE" ) if not res['jobDefinitions']: - docker_image = docker_image = self._find_container(job_wrapper).container_id + docker_image = self._find_container(job_wrapper).container_id # user_id = job_wrapper.user.id jd_arn = self._register_job_definition(jd_name, docker_image, destination_params) else: From 23bdcd9d69644abdd9a98c924a1fa879f768f2f8 Mon Sep 17 00:00:00 2001 From: qiagu Date: Fri, 4 Mar 2022 15:06:50 -0800 Subject: [PATCH 07/43] Remove jobs from from watched if finished or failed --- lib/galaxy/jobs/runners/aws.py | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index fa2445b9950..5c4f420097e 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -350,9 +350,11 @@ class AWSBatchJobRunner(AsynchronousJobRunner): @handle_exception_call def check_watched_items(self): - self.check_watched_items_by_batch(0, len(self.watched)) + done = [] + self.check_watched_items_by_batch(0, len(self.watched), done) + self.watched = [x for x in self.watched if x[0] not in done] - def check_watched_items_by_batch(self, start: int, end: int): + def check_watched_items_by_batch(self, start: int, end: int, done: list): jobs = self.watched[start: start+self.MAX_JOBS_PER_QUERY] if not jobs: return @@ -369,26 +371,23 @@ class AWSBatchJobRunner(AsynchronousJobRunner): job_state = jobs_dict[job_id] if status == 'SUCCEEDED': - self.watched.remove((job_id, job_state)) - start -= 1 self._mark_as_successful(job_state) + done.append(job_id) elif status == 'FAILED': - self.watched.remove((job_id, job_state)) - start -= 1 reason = job['statusReason'] self._mark_as_failed(job_state, reason) + done.append(job_id) elif status in ('SUBMITTED', 'PENDING', 'RUNNABLE', 'STARTING', 'RUNNING'): self._mark_as_active(job_state) # TODO else? for job_id in set(jobs_dict.keys()) - set(gotten): job_state = jobs_dict[job_id] - self.watched.remove((job_id, job_state)) - start -= 1 reason = f"The track of Job {job_state} was lost for unknown reason!" self._mark_as_failed(job_state, reason) + done.append(job_id) - self.check_watched_items_by_batch(start+self.MAX_JOBS_PER_QUERY, end) + self.check_watched_items_by_batch(start+self.MAX_JOBS_PER_QUERY, end, done) def _mark_as_successful(self, job_state): msg = "Job {name!r} finished successfully" From 62128d54f88a4e943ef32b85f1dcdcc8a198e8e0 Mon Sep 17 00:00:00 2001 From: qiagu Date: Mon, 7 Mar 2022 17:34:43 -0800 Subject: [PATCH 08/43] Add EC2 and GPU container support --- lib/galaxy/jobs/runners/aws.py | 172 +++++++++++++++++---------------- 1 file changed, 91 insertions(+), 81 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 5c4f420097e..20c8546584f 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -45,6 +45,7 @@ def handle_exception_call(func): return func(*args, **kwargs) except Exception as e: LOGGER.error(unicodify(e)) + raise return wrapper @@ -60,12 +61,27 @@ def _write_logfile(logfile, msg): fil.write(msg) -def _add_galaxy_environment_variables(cpus, memory): +def _add_galaxy_environment_variables(vcpu, memory): # Set: # GALAXY_SLOTS, round 0.25 vpuc to 1. # GALAXY_MEMORY_MB - return [{"name": "GALAXY_SLOTS", "value": str(int(max(cpus, 1)))}, - {"name": "GALAXY_MEMORY_MB", "value": str(memory)}] + return [ + {"name": "GALAXY_SLOTS", "value": str(int(max(vcpu, 1)))}, + {"name": "GALAXY_MEMORY_MB", "value": str(memory)} + ] + + +def _add_resource_requirements(destination_params): + rval = [ + {'type': 'VCPU', 'value': str(destination_params.get('vcpu'))}, + {'type': 'MEMORY', 'value': str(destination_params.get('memory'))} + ] + n_gpu = destination_params.get('gpu') + if n_gpu: + rval.append( + {'type': 'GPU', 'value': str(n_gpu)} + ) + return rval class AWSBatchJobRunner(AsynchronousJobRunner): @@ -90,35 +106,40 @@ class AWSBatchJobRunner(AsynchronousJobRunner): "vcpu": { "default": 1.0, "map_name": "vcpu", - "map": float, + "map": (lambda x: int(float(x)) if int(float(x))==float(x) else float(x)), }, "memory": { "default": 2048, "map_name": "memory", "map": int, }, + "gpu": { + "default": 0, + "map_name": "gpu", + "map": int, + }, "job_queue": { - "default": None, + "default": '', "map_name": "job_queue", "map": str, }, "job_role_arn": { - "default": None, + "default": '', "map_name": "job_role_arn", "map": str, }, "efs_filesystem_id": { - "default": None, + "default": '', "map_name": "efs_filesystem_id", "map": str, }, "efs_mount_point": { - "default": None, + "default": '', "map_name": "efs_mount_point", "map": str, }, "fargate_version": { - "default": None, + "default": '', "map_name": "fargate_version", "map": str, } @@ -145,16 +166,9 @@ class AWSBatchJobRunner(AsynchronousJobRunner): return job_destination = job_wrapper.job_destination - if not job_destination.params.get("docker_enabled"): - raise AWSBatchRunnerException("AWSBatchJobRunner needs 'docker_enabled' to be set as True") - destination_params = self.parse_destination_params(job_destination.params) job_def = self._get_job_definition(job_wrapper, destination_params) - if 'fargate_version' in destination_params: - job_name, job_id = self._submit_fargate_job(job_def, job_wrapper, destination_params) - else: - job_name, job_id = self._submit_ec2_job(job_def, job_wrapper, destination_params) - + job_name, job_id = self._submit_job(job_def, job_wrapper, destination_params) job_wrapper.set_external_id(job_id) ajs = AsynchronousJobState( files_dir=job_wrapper.working_directory, @@ -186,80 +200,76 @@ class AWSBatchJobRunner(AsynchronousJobRunner): jd_arn = self._register_job_definition(jd_name, docker_image, destination_params) else: jd_arn = res['jobDefinitions'][0]['jobDefinitionArn'] + LOGGER.debug(f"Found existing job definition: {jd_name}.") return jd_arn def _register_job_definition(self, jd_name, docker_image, destination_params): - if 'fargate_version' in destination_params: - return self._register_fargate_job_definition(jd_name, docker_image, destination_params) - else: - return self._register_ec2_job_definition(jd_name, docker_image, destination_params) + LOGGER.debug(f"Registering a new job definition: {jd_name}.") + platform = 'FARGATE' if destination_params.get('fargate_version') else 'EC2' + # TODO: support multi-node + containerProperties = { + 'image': docker_image, + 'command': [ + '/bin/sh', + ], + 'jobRoleArn': destination_params.get('job_role_arn'), + 'executionRoleArn': destination_params.get('execute_role_arn', '') or destination_params.get('job_role_arn'), + 'volumes': [ + { + 'name': 'efs_whole', + 'efsVolumeConfiguration': { + 'fileSystemId': destination_params.get('efs_filesystem_id'), + 'rootDirectory': '/', + 'transitEncryption': 'ENABLED', + 'authorizationConfig': { + 'iam': 'ENABLED' + } + } + }, + ], + 'mountPoints': [ + { + 'containerPath': destination_params.get('efs_mount_point'), + 'readOnly': False, + 'sourceVolume': 'efs_whole' + }, + ], + 'resourceRequirements': _add_resource_requirements(destination_params), + 'environment': _add_galaxy_environment_variables( + destination_params.get('vcpu'), + destination_params.get('memory') + ), + 'logConfiguration': { + 'logDriver': 'awslogs' + } + } + if platform == 'FARGATE': + containerProperties.update( + { + 'networkConfiguration': { + 'assignPublicIp': 'ENABLED' + }, + 'fargatePlatformConfiguration': { + 'platformVersion': destination_params.get('fargate_version') + }, + 'logConfiguration': { + 'logDriver': 'awslogs' + } + } + ) - def _register_fargate_job_definition(self, jd_name, docker_image, destination_params): res = self._batch_client.register_job_definition( jobDefinitionName=jd_name, type='container', - platformCapabilities=['FARGATE'], - containerProperties={ - 'image': docker_image, - 'command': [ - '/bin/sh', - ], - 'jobRoleArn': destination_params.get('job_role_arn'), - 'executionRoleArn': destination_params.get('execute_role_arn', None) or destination_params.get('job_role_arn'), - 'volumes': [ - { - 'name': 'efs_whole', - 'efsVolumeConfiguration': { - 'fileSystemId': destination_params.get('efs_filesystem_id'), - 'rootDirectory': '/', - 'transitEncryption': 'ENABLED', - 'authorizationConfig': { - 'iam': 'ENABLED' - } - } - }, - ], - 'mountPoints': [ - { - 'containerPath': destination_params.get('efs_mount_point'), - 'readOnly': False, - 'sourceVolume': 'efs_whole' - }, - ], - 'resourceRequirements': [ - { - 'type': 'VCPU', - 'value': str(destination_params.get('vcpu')) - }, - { - 'type': 'MEMORY', - 'value': str(destination_params.get('memory')) - } - ], - 'environment': _add_galaxy_environment_variables( - destination_params.get('vcpu'), - destination_params.get('memory') - ), - 'logConfiguration': { - 'logDriver': 'awslogs' - }, - 'networkConfiguration': { - 'assignPublicIp': 'ENABLED' - }, - 'fargatePlatformConfiguration': { - 'platformVersion': destination_params.get('fargate_version') - } - } + platformCapabilities=[platform], + containerProperties=containerProperties ) assert res['ResponseMetadata']['HTTPStatusCode'] == 200 return res['jobDefinitionArn'] - def _register_ec2_job_definition(self, jd_name, docker_image, user_id, destination_params): - return '' - - def _submit_fargate_job(self, job_def, job_wrapper, destination_params): + def _submit_job(self, job_def, job_wrapper, destination_params): job_name = self.JOB_NAME_PREFIX + job_wrapper.get_id_tag() command_script_path = self.write_command(job_wrapper) @@ -278,9 +288,6 @@ class AWSBatchJobRunner(AsynchronousJobRunner): assert res['ResponseMetadata']['HTTPStatusCode'] == 200 return job_name, res['jobId'] - def _submit_ec2_job(self, job_def, job_wrapper, destination_params): - pass - @handle_exception_call def stop_job(self, job_wrapper): job = job_wrapper.get_job() @@ -415,6 +422,9 @@ class AWSBatchJobRunner(AsynchronousJobRunner): # self._batch_client.delete(job_state.job_id) def parse_destination_params(self, params): + if not params.get("docker_enabled"): + raise AWSBatchRunnerException("AWSBatchJobRunner needs 'docker_enabled' to be set as True") + parsed_params = {} for k, spec in self.DESTINATION_PARAMS_SPEC.items(): value = params.get(k, spec.get("default")) From 786123d1f0f0aca7089c77941afb5d56af4fbeee Mon Sep 17 00:00:00 2001 From: qiagu Date: Tue, 8 Mar 2022 19:11:44 -0800 Subject: [PATCH 09/43] Enable auto platform --- lib/galaxy/jobs/runners/aws.py | 73 +++++++++++++++++++++++++++++----- 1 file changed, 64 insertions(+), 9 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 20c8546584f..73d5a470198 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -142,9 +142,25 @@ class AWSBatchJobRunner(AsynchronousJobRunner): "default": '', "map_name": "fargate_version", "map": str, + }, + "auto_platform": { + "default": False, + "map_name": "auto_platform", + "map": lambda x: x in ["true", "True", "TRUE"] } } + FARGATE_VCPUS = [0.25, 0.5, 1, 2, 4] + + FARGATE_RESOURCES = { + 0.25: [512, 1024, 2048], + 0.5: [1024, 2048, 3072, 4096], + 1: [2048, 3072, 4096, 5120, 6144, 7168, 8192], + 2: [4096, 5120, 6144, 7168, 8192, 9216, 10240, 11264, 12288, 13312, 14336, 15360, 16384], + 4: [8192, 9216, 10240, 11264, 12288, 13312, 14336, 15360, 16384, 17408, 18432, + 19456, 20480, 21504, 22528, 23552, 24576, 25600, 26624, 27648, 28672, 29696, 30720] + } + def __init__(self, app, nworkers, **kwargs): """Initialize this job runner and start the monitor thread""" assert boto3, BOTO3_IMPORT_MSG @@ -196,7 +212,6 @@ class AWSBatchJobRunner(AsynchronousJobRunner): ) if not res['jobDefinitions']: docker_image = self._find_container(job_wrapper).container_id - # user_id = job_wrapper.user.id jd_arn = self._register_job_definition(jd_name, docker_image, destination_params) else: jd_arn = res['jobDefinitions'][0]['jobDefinitionArn'] @@ -206,7 +221,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): def _register_job_definition(self, jd_name, docker_image, destination_params): LOGGER.debug(f"Registering a new job definition: {jd_name}.") - platform = 'FARGATE' if destination_params.get('fargate_version') else 'EC2' + platform = destination_params.get('platform') # TODO: support multi-node containerProperties = { 'image': docker_image, @@ -214,7 +229,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): '/bin/sh', ], 'jobRoleArn': destination_params.get('job_role_arn'), - 'executionRoleArn': destination_params.get('execute_role_arn', '') or destination_params.get('job_role_arn'), + 'executionRoleArn': destination_params.get('execute_role_arn') or destination_params.get('job_role_arn'), 'volumes': [ { 'name': 'efs_whole', @@ -416,14 +431,9 @@ class AWSBatchJobRunner(AsynchronousJobRunner): job_state.fail_message = reason self.mark_as_failed(job_state) - # @handle_exception_call - # def finish_job(self, job_state): - # super().finish_job(job_state) - # self._batch_client.delete(job_state.job_id) - def parse_destination_params(self, params): if not params.get("docker_enabled"): - raise AWSBatchRunnerException("AWSBatchJobRunner needs 'docker_enabled' to be set as True") + raise AWSBatchRunnerException("AWSBatchJobRunner needs 'docker_enabled' to be set as True!") parsed_params = {} for k, spec in self.DESTINATION_PARAMS_SPEC.items(): @@ -432,6 +442,51 @@ class AWSBatchJobRunner(AsynchronousJobRunner): mapper = spec.get("map") segments = map_to.split("/") parsed_params.update(to_dict(segments, mapper(value))) + + # parse Platform + platform = 'EC2' + auto_platform = parsed_params.get('auto_platform') + fargate_version = parsed_params.get('fargate_version') + vcpu = parsed_params.get('vcpu') + memory = parsed_params.get('memory') + gpu = parsed_params.get('gpu') + + if auto_platform and not fargate_version: + raise AWSBatchRunnerException("AWSBatchJobRunner needs 'farget_version' to be set to enable auto platform!") + + if gpu and (fargate_version or auto_platform): + raise AWSBatchRunnerException("GPU mode is not allowed when 'fargate_version' and/or 'auto_platform' are set!") + + if fargate_version and not auto_platform: + platform = 'FARGATE' + if auto_platform: + max_vcpu = self.FARGATE_VCPUS[-1] + max_memory = self.FARGATE_RESOURCES[max_vcpu][-1] + if vcpu <= max_vcpu and memory <= max_memory: + new_vcpu, new_memory = None, None + for c in self.FARGATE_VCPUS: + if c < vcpu: + continue + for m in self.FARGATE_RESOURCES[c]: + if m >= memory: #Found the best match + new_vcpu = c + new_memory = m + break + if new_memory: + platform = 'FARGATE' + parsed_params['vcpu'] = new_vcpu + parsed_params['memory'] = new_memory + break + # parse JOB QUEUE + job_queues = parsed_params.get('job_queue').split(',') + if len(job_queues) < 2: + raise AWSBatchRunnerException( + "AWSBatchJobRunner needs TWO job queues ('Farget Queue, EC2 Qeueue')" + " when 'auto_platform' is enabled!" + ) + parsed_params['job_queue'] = job_queues[platform == 'EC2'].strip() + + parsed_params['platform'] = platform return parsed_params def write_command(self, job_wrapper): From 1734db39a2e3f740911b029c7882ceacc21969d9 Mon Sep 17 00:00:00 2001 From: qiagu Date: Wed, 9 Mar 2022 16:03:11 -0800 Subject: [PATCH 10/43] Catch stdout/stderr using AWS CloudWatchLogs --- lib/galaxy/jobs/runners/aws.py | 72 ++++++++++++++++++++++------------ 1 file changed, 48 insertions(+), 24 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 73d5a470198..62a61dfa2cb 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -95,57 +95,61 @@ class AWSBatchJobRunner(AsynchronousJobRunner): RUNNER_PARAM_SPEC = { "aws_access_key_id": { - "map": str, + "map": str }, "aws_secret_access_key": { - "map": str, + "map": str } } DESTINATION_PARAMS_SPEC = { "vcpu": { "default": 1.0, - "map_name": "vcpu", "map": (lambda x: int(float(x)) if int(float(x))==float(x) else float(x)), }, "memory": { "default": 2048, - "map_name": "memory", "map": int, }, "gpu": { "default": 0, - "map_name": "gpu", "map": int, }, "job_queue": { - "default": '', - "map_name": "job_queue", + "default": None, "map": str, + "required": True }, "job_role_arn": { - "default": '', - "map_name": "job_role_arn", + "default": None, "map": str, + "required": True }, "efs_filesystem_id": { - "default": '', - "map_name": "efs_filesystem_id", + "default": None, "map": str, + "required": True }, "efs_mount_point": { + "default": None, + "map": str, + "required": True + }, + "log_group_name": { + "default": None, + "map": str, + "required": True + }, + "execute_role_arn": { "default": '', - "map_name": "efs_mount_point", "map": str, }, "fargate_version": { "default": '', - "map_name": "fargate_version", "map": str, }, "auto_platform": { "default": False, - "map_name": "auto_platform", "map": lambda x: x in ["true", "True", "TRUE"] } } @@ -173,6 +177,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): aws_secret_access_key=self.runner_params.get('aws_secret_access_key') or None ) self._batch_client = session.client('batch') + self._logs_client = session.client('logs') @handle_exception_call def queue_job(self, job_wrapper): @@ -389,15 +394,18 @@ class AWSBatchJobRunner(AsynchronousJobRunner): for job in res['jobs']: status = job['status'] job_id = job['jobId'] + log_stream_name = job['container']['logStreamName'] gotten.append(job_id) job_state = jobs_dict[job_id] if status == 'SUCCEEDED': - self._mark_as_successful(job_state) + logs = self._get_log_events(job_state, log_stream_name) + self._mark_as_successful(job_state, logs) done.append(job_id) elif status == 'FAILED': + logs = self._get_log_events(job_state, log_stream_name) reason = job['statusReason'] - self._mark_as_failed(job_state, reason) + self._mark_as_failed(job_state, reason, logs) done.append(job_id) elif status in ('SUBMITTED', 'PENDING', 'RUNNABLE', 'STARTING', 'RUNNING'): self._mark_as_active(job_state) @@ -411,9 +419,17 @@ class AWSBatchJobRunner(AsynchronousJobRunner): self.check_watched_items_by_batch(start+self.MAX_JOBS_PER_QUERY, end, done) - def _mark_as_successful(self, job_state): - msg = "Job {name!r} finished successfully" - _write_logfile(job_state.output_file, msg.format(name=job_state.job_name)) + def _get_log_events(self, job_state, log_stream_name): + log_group_name = job_state.job_destination.params.get('log_group_name') + res = self._logs_client.get_log_events( + logGroupName=log_group_name, + logStreamName=log_stream_name + ) + messages = [e['message']for e in res['events']] + return '\n'.join(messages) + + def _mark_as_successful(self, job_state, logs): + _write_logfile(job_state.output_file, logs) _write_logfile(job_state.error_file, "") job_state.running = False job_state.job_wrapper.change_state(model.Job.states.OK) @@ -423,7 +439,9 @@ class AWSBatchJobRunner(AsynchronousJobRunner): job_state.running = True job_state.job_wrapper.change_state(model.Job.states.RUNNING) - def _mark_as_failed(self, job_state, reason): + def _mark_as_failed(self, job_state, reason, logs): + if logs: + reason = '\n\n'.join((reason, logs)) _write_logfile(job_state.error_file, reason) job_state.running = False job_state.stop_job = False @@ -435,13 +453,17 @@ class AWSBatchJobRunner(AsynchronousJobRunner): if not params.get("docker_enabled"): raise AWSBatchRunnerException("AWSBatchJobRunner needs 'docker_enabled' to be set as True!") + check_required = [] parsed_params = {} for k, spec in self.DESTINATION_PARAMS_SPEC.items(): value = params.get(k, spec.get("default")) - map_to = spec.get("map_name") + if spec.get('required') and not value: + check_required.append(k) mapper = spec.get("map") - segments = map_to.split("/") - parsed_params.update(to_dict(segments, mapper(value))) + parsed_params[k] = mapper(value) + if check_required: + raise AWSBatchRunnerException("AWSBatchJobRunner requires the following params to be provided: %s." + % (', '.join(check_required))) # parse Platform platform = 'EC2' @@ -498,8 +520,10 @@ class AWSBatchJobRunner(AsynchronousJobRunner): path = f"{job_wrapper.working_directory}/galaxy_{job_wrapper.get_id_tag()}.sh" mode = 0o755 + runner_command_line = job_wrapper.runner_command_line.replace( + '> ../outputs/tool_stdout 2> ../outputs/tool_stderr', '') with open(path, "w", encoding="utf-8") as f: f.write("#!/bin/bash\n") - f.write(job_wrapper.runner_command_line) + f.write(runner_command_line) os.chmod(path, mode) return path From 704d60c94566df78dd40d0a4dc3e4fb2fa5a83c1 Mon Sep 17 00:00:00 2001 From: qiagu Date: Wed, 9 Mar 2022 17:03:51 -0800 Subject: [PATCH 11/43] Catch get log events error --- lib/galaxy/jobs/runners/aws.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 62a61dfa2cb..4a90ad40ee2 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -421,12 +421,15 @@ class AWSBatchJobRunner(AsynchronousJobRunner): def _get_log_events(self, job_state, log_stream_name): log_group_name = job_state.job_destination.params.get('log_group_name') - res = self._logs_client.get_log_events( - logGroupName=log_group_name, - logStreamName=log_stream_name - ) - messages = [e['message']for e in res['events']] - return '\n'.join(messages) + try: + res = self._logs_client.get_log_events( + logGroupName=log_group_name, + logStreamName=log_stream_name + ) + messages = [e['message']for e in res['events']] + return '\n'.join(messages) + except Exception as e: + LOGGER.error(e) def _mark_as_successful(self, job_state, logs): _write_logfile(job_state.output_file, logs) From 6457d9af0b0ad8decea313084a7190e5b813eb35 Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 10 Mar 2022 09:38:14 -0800 Subject: [PATCH 12/43] Remove cloudwatch logs and add container user --- lib/galaxy/jobs/runners/aws.py | 44 +++++++++------------------------- 1 file changed, 11 insertions(+), 33 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 4a90ad40ee2..3861740f9be 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -135,11 +135,6 @@ class AWSBatchJobRunner(AsynchronousJobRunner): "map": str, "required": True }, - "log_group_name": { - "default": None, - "map": str, - "required": True - }, "execute_role_arn": { "default": '', "map": str, @@ -177,7 +172,6 @@ class AWSBatchJobRunner(AsynchronousJobRunner): aws_secret_access_key=self.runner_params.get('aws_secret_access_key') or None ) self._batch_client = session.client('batch') - self._logs_client = session.client('logs') @handle_exception_call def queue_job(self, job_wrapper): @@ -260,6 +254,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): destination_params.get('vcpu'), destination_params.get('memory') ), + 'user': '%d:%d' % (os.getuid(), os.getgid()), 'logConfiguration': { 'logDriver': 'awslogs' } @@ -394,18 +389,15 @@ class AWSBatchJobRunner(AsynchronousJobRunner): for job in res['jobs']: status = job['status'] job_id = job['jobId'] - log_stream_name = job['container']['logStreamName'] gotten.append(job_id) job_state = jobs_dict[job_id] if status == 'SUCCEEDED': - logs = self._get_log_events(job_state, log_stream_name) - self._mark_as_successful(job_state, logs) + self._mark_as_successful(job_state) done.append(job_id) elif status == 'FAILED': - logs = self._get_log_events(job_state, log_stream_name) reason = job['statusReason'] - self._mark_as_failed(job_state, reason, logs) + self._mark_as_failed(job_state, reason) done.append(job_id) elif status in ('SUBMITTED', 'PENDING', 'RUNNABLE', 'STARTING', 'RUNNING'): self._mark_as_active(job_state) @@ -419,20 +411,8 @@ class AWSBatchJobRunner(AsynchronousJobRunner): self.check_watched_items_by_batch(start+self.MAX_JOBS_PER_QUERY, end, done) - def _get_log_events(self, job_state, log_stream_name): - log_group_name = job_state.job_destination.params.get('log_group_name') - try: - res = self._logs_client.get_log_events( - logGroupName=log_group_name, - logStreamName=log_stream_name - ) - messages = [e['message']for e in res['events']] - return '\n'.join(messages) - except Exception as e: - LOGGER.error(e) - - def _mark_as_successful(self, job_state, logs): - _write_logfile(job_state.output_file, logs) + def _mark_as_successful(self, job_state): + _write_logfile(job_state.output_file, "") _write_logfile(job_state.error_file, "") job_state.running = False job_state.job_wrapper.change_state(model.Job.states.OK) @@ -442,9 +422,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): job_state.running = True job_state.job_wrapper.change_state(model.Job.states.RUNNING) - def _mark_as_failed(self, job_state, reason, logs): - if logs: - reason = '\n\n'.join((reason, logs)) + def _mark_as_failed(self, job_state, reason): _write_logfile(job_state.error_file, reason) job_state.running = False job_state.stop_job = False @@ -517,16 +495,16 @@ class AWSBatchJobRunner(AsynchronousJobRunner): def write_command(self, job_wrapper): # Create command script instead passing it in the container # preventing wrong characters parsing. - if not os.path.exists(job_wrapper.working_directory): + job_directory = job_wrapper.working_directory + if not os.path.exists(job_directory): LOGGER.error("No working directory found") - path = f"{job_wrapper.working_directory}/galaxy_{job_wrapper.get_id_tag()}.sh" + path = f"{job_directory}/galaxy_{job_wrapper.get_id_tag()}.sh" mode = 0o755 - runner_command_line = job_wrapper.runner_command_line.replace( - '> ../outputs/tool_stdout 2> ../outputs/tool_stderr', '') with open(path, "w", encoding="utf-8") as f: f.write("#!/bin/bash\n") - f.write(runner_command_line) + f.write(f"cd {job_directory}") + f.write(job_wrapper.runner_command_line) os.chmod(path, mode) return path From bb1e79004c22e922e873f8e1821fcf3c906a0138 Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 10 Mar 2022 10:39:47 -0800 Subject: [PATCH 13/43] Modernize job command file --- lib/galaxy/jobs/runners/aws.py | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 3861740f9be..9177ccd7bbb 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -5,13 +5,17 @@ import os import time from queue import Empty +from . import JobState from galaxy import model +from galaxy.job_execution.output_collect import default_exit_code_file from galaxy.jobs.runners import ( AsynchronousJobRunner, AsynchronousJobState, ) from galaxy.util import smart_str, unicodify + + BOTO3_IMPORT_MSG = ( "The Python 'boto3' package is required to use " "this feature, please install it or correct the " @@ -495,16 +499,16 @@ class AWSBatchJobRunner(AsynchronousJobRunner): def write_command(self, job_wrapper): # Create command script instead passing it in the container # preventing wrong characters parsing. - job_directory = job_wrapper.working_directory - if not os.path.exists(job_directory): - LOGGER.error("No working directory found") - - path = f"{job_directory}/galaxy_{job_wrapper.get_id_tag()}.sh" - mode = 0o755 - - with open(path, "w", encoding="utf-8") as f: - f.write("#!/bin/bash\n") - f.write(f"cd {job_directory}") - f.write(job_wrapper.runner_command_line) - os.chmod(path, mode) - return path + command_line = job_wrapper.runner_command_line + job_id = job_wrapper.get_id_tag() + job_file = JobState.default_job_file(job_wrapper.working_directory, job_id) + exit_code_path = default_exit_code_file(job_wrapper.working_directory, job_id) + job_script_props = { + "command": command_line, + "exit_code_path": exit_code_path, + "working_directory": job_wrapper.working_directory, + "shell": job_wrapper.shell, + } + job_file_contents = self.get_job_file(job_wrapper, **job_script_props) + self.write_executable_script(job_file, job_file_contents, job_io=job_wrapper.job_io) + return job_file From f92aed4e6040c114564b3c1f4a9960d41a8def14 Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 10 Mar 2022 10:48:24 -0800 Subject: [PATCH 14/43] Fix linter issues --- lib/galaxy/jobs/runners/aws.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 9177ccd7bbb..7d12b8e23b8 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -15,7 +15,6 @@ from galaxy.jobs.runners import ( from galaxy.util import smart_str, unicodify - BOTO3_IMPORT_MSG = ( "The Python 'boto3' package is required to use " "this feature, please install it or correct the " @@ -41,7 +40,7 @@ class AWSBatchRunnerException(Exception): def handle_exception_call(func): - # Catch boto3 exceptions. + # Catch boto3 exceptions. @functools.wraps(func) def wrapper(*args, **kwargs): @@ -109,7 +108,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): DESTINATION_PARAMS_SPEC = { "vcpu": { "default": 1.0, - "map": (lambda x: int(float(x)) if int(float(x))==float(x) else float(x)), + "map": (lambda x: int(float(x)) if int(float(x)) == float(x) else float(x)), }, "memory": { "default": 2048, @@ -205,7 +204,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): h.update(smart_str(tool_id)) h.update(smart_str(tool_version)) for k, v in destination_params.items(): - h.update(smart_str(k+str(v))) + h.update(smart_str(k + str(v))) queue_name = destination_params.get('job_queue').rsplit('/', 1)[-1] jd_name = f"galaxy_tool__{tool_id}__{h.hexdigest()}__{queue_name}" @@ -219,7 +218,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): else: jd_arn = res['jobDefinitions'][0]['jobDefinitionArn'] LOGGER.debug(f"Found existing job definition: {jd_name}.") - + return jd_arn def _register_job_definition(self, jd_name, docker_image, destination_params): @@ -381,7 +380,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): self.watched = [x for x in self.watched if x[0] not in done] def check_watched_items_by_batch(self, start: int, end: int, done: list): - jobs = self.watched[start: start+self.MAX_JOBS_PER_QUERY] + jobs = self.watched[start: start + self.MAX_JOBS_PER_QUERY] if not jobs: return @@ -413,7 +412,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): self._mark_as_failed(job_state, reason) done.append(job_id) - self.check_watched_items_by_batch(start+self.MAX_JOBS_PER_QUERY, end, done) + self.check_watched_items_by_batch(start + self.MAX_JOBS_PER_QUERY, end, done) def _mark_as_successful(self, job_state): _write_logfile(job_state.output_file, "") @@ -475,7 +474,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): if c < vcpu: continue for m in self.FARGATE_RESOURCES[c]: - if m >= memory: #Found the best match + if m >= memory: # Found the best match new_vcpu = c new_memory = m break From a137685f9aaf4086887600155a6118fb86e7e596 Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 10 Mar 2022 11:23:36 -0800 Subject: [PATCH 15/43] Hash container image in job definition --- lib/galaxy/jobs/runners/aws.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 7d12b8e23b8..9828649c23e 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -199,10 +199,9 @@ class AWSBatchJobRunner(AsynchronousJobRunner): def _get_job_definition(self, job_wrapper, destination_params): tool_id = job_wrapper.tool.id - tool_version = job_wrapper.tool.version + container_image = self._find_container(job_wrapper).container_id h = hashlib.new("sha256") - h.update(smart_str(tool_id)) - h.update(smart_str(tool_version)) + h.update(smart_str(container_image)) for k, v in destination_params.items(): h.update(smart_str(k + str(v))) queue_name = destination_params.get('job_queue').rsplit('/', 1)[-1] @@ -213,20 +212,19 @@ class AWSBatchJobRunner(AsynchronousJobRunner): status="ACTIVE" ) if not res['jobDefinitions']: - docker_image = self._find_container(job_wrapper).container_id - jd_arn = self._register_job_definition(jd_name, docker_image, destination_params) + jd_arn = self._register_job_definition(jd_name, container_image, destination_params) else: jd_arn = res['jobDefinitions'][0]['jobDefinitionArn'] LOGGER.debug(f"Found existing job definition: {jd_name}.") return jd_arn - def _register_job_definition(self, jd_name, docker_image, destination_params): + def _register_job_definition(self, jd_name, container_image, destination_params): LOGGER.debug(f"Registering a new job definition: {jd_name}.") platform = destination_params.get('platform') # TODO: support multi-node containerProperties = { - 'image': docker_image, + 'image': container_image, 'command': [ '/bin/sh', ], @@ -291,6 +289,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): job_name = self.JOB_NAME_PREFIX + job_wrapper.get_id_tag() command_script_path = self.write_command(job_wrapper) + LOGGER.info(f"Submitting job {job_name} to AWS Batch.") res = self._batch_client.submit_job( jobName=job_name, jobQueue=destination_params.get('job_queue'), From 4413f3555dfdf9379cf9a45ecb5adced6793cd3f Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 10 Mar 2022 14:42:06 -0800 Subject: [PATCH 16/43] Fix linter errors --- lib/galaxy/jobs/runners/aws.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 9828649c23e..cdbae3a6513 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -108,15 +108,15 @@ class AWSBatchJobRunner(AsynchronousJobRunner): DESTINATION_PARAMS_SPEC = { "vcpu": { "default": 1.0, - "map": (lambda x: int(float(x)) if int(float(x)) == float(x) else float(x)), + "map": (lambda x: int(float(x)) if int(float(x)) == float(x) else float(x)) }, "memory": { "default": 2048, - "map": int, + "map": int }, "gpu": { "default": 0, - "map": int, + "map": int }, "job_queue": { "default": None, @@ -140,11 +140,11 @@ class AWSBatchJobRunner(AsynchronousJobRunner): }, "execute_role_arn": { "default": '', - "map": str, + "map": str }, "fargate_version": { "default": '', - "map": str, + "map": str }, "auto_platform": { "default": False, @@ -322,7 +322,6 @@ class AWSBatchJobRunner(AsynchronousJobRunner): ajs = AsynchronousJobState(files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper) ajs.job_id = str(job_id) ajs.job_name = job_name - ajs.command_line = job.command_line ajs.job_wrapper = job_wrapper ajs.job_destination = job_wrapper.job_destination if job.state in (model.Job.states.RUNNING, model.Job.states.STOPPED): @@ -378,7 +377,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): self.check_watched_items_by_batch(0, len(self.watched), done) self.watched = [x for x in self.watched if x[0] not in done] - def check_watched_items_by_batch(self, start: int, end: int, done: list): + def check_watched_items_by_batch(self, start: int, end: int, done: list[str]): jobs = self.watched[start: start + self.MAX_JOBS_PER_QUERY] if not jobs: return From dbbb8df41a87a16c2470a814c0be1c690ee0537a Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 10 Mar 2022 17:09:54 -0800 Subject: [PATCH 17/43] Fix typing errors --- lib/galaxy/jobs/runners/aws.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index cdbae3a6513..7d2ec88794d 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -5,6 +5,7 @@ import os import time from queue import Empty +from typing import List from . import JobState from galaxy import model from galaxy.job_execution.output_collect import default_exit_code_file @@ -373,11 +374,11 @@ class AWSBatchJobRunner(AsynchronousJobRunner): @handle_exception_call def check_watched_items(self): - done = [] + done = [] # type: ignore self.check_watched_items_by_batch(0, len(self.watched), done) self.watched = [x for x in self.watched if x[0] not in done] - def check_watched_items_by_batch(self, start: int, end: int, done: list[str]): + def check_watched_items_by_batch(self, start: int, end: int, done: List[str]): jobs = self.watched[start: start + self.MAX_JOBS_PER_QUERY] if not jobs: return @@ -438,10 +439,10 @@ class AWSBatchJobRunner(AsynchronousJobRunner): check_required = [] parsed_params = {} for k, spec in self.DESTINATION_PARAMS_SPEC.items(): - value = params.get(k, spec.get("default")) - if spec.get('required') and not value: + value = params.get(k, spec.get("default")) # type: ignore[attr-defined] + if spec.get('required') and not value: # type: ignore[attr-defined] check_required.append(k) - mapper = spec.get("map") + mapper = spec.get("map") # type: ignore[attr-defined] parsed_params[k] = mapper(value) if check_required: raise AWSBatchRunnerException("AWSBatchJobRunner requires the following params to be provided: %s." @@ -466,13 +467,13 @@ class AWSBatchJobRunner(AsynchronousJobRunner): if auto_platform: max_vcpu = self.FARGATE_VCPUS[-1] max_memory = self.FARGATE_RESOURCES[max_vcpu][-1] - if vcpu <= max_vcpu and memory <= max_memory: + if vcpu <= max_vcpu and memory <= max_memory: # type: ignore[operator] new_vcpu, new_memory = None, None for c in self.FARGATE_VCPUS: - if c < vcpu: + if c < vcpu: # type: ignore[operator] continue for m in self.FARGATE_RESOURCES[c]: - if m >= memory: # Found the best match + if m >= memory: # type: ignore[operator] new_vcpu = c new_memory = m break @@ -482,7 +483,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner): parsed_params['memory'] = new_memory break # parse JOB QUEUE - job_queues = parsed_params.get('job_queue').split(',') + job_queues = parsed_params.get('job_queue').split(',') # type: ignore[union-attr] if len(job_queues) < 2: raise AWSBatchRunnerException( "AWSBatchJobRunner needs TWO job queues ('Farget Queue, EC2 Qeueue')" From 0466da84416aa7320ea109c3510bc7a40758270e Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 10 Mar 2022 17:27:14 -0800 Subject: [PATCH 18/43] Add job config sample for AWSBatchJobRunner --- .../sample/job_conf.xml.sample_advanced | 29 ++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/lib/galaxy/config/sample/job_conf.xml.sample_advanced b/lib/galaxy/config/sample/job_conf.xml.sample_advanced index 9dab3586f57..58043830f41 100644 --- a/lib/galaxy/config/sample/job_conf.xml.sample_advanced +++ b/lib/galaxy/config/sample/job_conf.xml.sample_advanced @@ -369,6 +369,11 @@ false + + + + - + + + true + arn_for_Fargate_job_queue, arn_for_EC2_job_queue + arn:aws:iam::xxxxxxxxxxxxxxxxxx + 1 + 2048 + fs-xxxxxxxxxxxxxx + /mnt/efs + 1.4.0 + true + + + true + + arn_for_gpu_job_queue + arn:aws:iam::xxxxxxxxxxxxxxxxxx + 4 + 20000 + 1 + fs-xxxxxxxxxxxxxx + /mnt/efs + From ac2d4bada0480034121dcf0523e91a1aca11e654 Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 10 Mar 2022 17:59:03 -0800 Subject: [PATCH 19/43] Fix black format issues --- lib/galaxy/jobs/runners/aws.py | 208 ++++++++++++++++----------------- 1 file changed, 99 insertions(+), 109 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 7d2ec88794d..15800354548 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -71,20 +71,18 @@ def _add_galaxy_environment_variables(vcpu, memory): # GALAXY_MEMORY_MB return [ {"name": "GALAXY_SLOTS", "value": str(int(max(vcpu, 1)))}, - {"name": "GALAXY_MEMORY_MB", "value": str(memory)} + {"name": "GALAXY_MEMORY_MB", "value": str(memory)}, ] def _add_resource_requirements(destination_params): rval = [ - {'type': 'VCPU', 'value': str(destination_params.get('vcpu'))}, - {'type': 'MEMORY', 'value': str(destination_params.get('memory'))} + {"type": "VCPU", "value": str(destination_params.get("vcpu"))}, + {"type": "MEMORY", "value": str(destination_params.get("memory"))}, ] - n_gpu = destination_params.get('gpu') + n_gpu = destination_params.get("gpu") if n_gpu: - rval.append( - {'type': 'GPU', 'value': str(n_gpu)} - ) + rval.append({"type": "GPU", "value": str(n_gpu)}) return rval @@ -97,64 +95,67 @@ class AWSBatchJobRunner(AsynchronousJobRunner): # Higher minimum interval as jobs are queried in batches. MIN_QUERY_INTERVAL = 10 + # fmt: off RUNNER_PARAM_SPEC = { "aws_access_key_id": { - "map": str + "map": str, }, "aws_secret_access_key": { - "map": str + "map": str, } } + # fmt: off DESTINATION_PARAMS_SPEC = { "vcpu": { "default": 1.0, - "map": (lambda x: int(float(x)) if int(float(x)) == float(x) else float(x)) + "map": (lambda x: int(float(x)) if int(float(x)) == float(x) else float(x)), }, "memory": { "default": 2048, - "map": int + "map": int, }, "gpu": { "default": 0, - "map": int + "map": int, }, "job_queue": { "default": None, "map": str, - "required": True + "required": True, }, "job_role_arn": { "default": None, "map": str, - "required": True + "required": True, }, "efs_filesystem_id": { "default": None, "map": str, - "required": True + "required": True, }, "efs_mount_point": { "default": None, "map": str, - "required": True + "required": True, }, "execute_role_arn": { - "default": '', - "map": str + "default": "", + "map": str, }, "fargate_version": { - "default": '', - "map": str + "default": "", + "map": str, }, "auto_platform": { "default": False, - "map": lambda x: x in ["true", "True", "TRUE"] + "map": lambda x: x in ["true", "True", "TRUE"], } } FARGATE_VCPUS = [0.25, 0.5, 1, 2, 4] + # fmt: off FARGATE_RESOURCES = { 0.25: [512, 1024, 2048], 0.5: [1024, 2048, 3072, 4096], @@ -172,10 +173,10 @@ class AWSBatchJobRunner(AsynchronousJobRunner): kwargs[self.RUNNER_PARAM_SPEC_KEY].update(self.RUNNER_PARAM_SPEC) super().__init__(app, nworkers, **kwargs) session = boto3.Session( - aws_access_key_id=self.runner_params.get('aws_access_key_id') or None, - aws_secret_access_key=self.runner_params.get('aws_secret_access_key') or None + aws_access_key_id=self.runner_params.get("aws_access_key_id") or None, + aws_secret_access_key=self.runner_params.get("aws_secret_access_key") or None ) - self._batch_client = session.client('batch') + self._batch_client = session.client("batch") @handle_exception_call def queue_job(self, job_wrapper): @@ -205,86 +206,72 @@ class AWSBatchJobRunner(AsynchronousJobRunner): h.update(smart_str(container_image)) for k, v in destination_params.items(): h.update(smart_str(k + str(v))) - queue_name = destination_params.get('job_queue').rsplit('/', 1)[-1] + queue_name = destination_params.get("job_queue").rsplit("/", 1)[-1] jd_name = f"galaxy_tool__{tool_id}__{h.hexdigest()}__{queue_name}" - res = self._batch_client.describe_job_definitions( - jobDefinitionName=jd_name, - status="ACTIVE" - ) - if not res['jobDefinitions']: + res = self._batch_client.describe_job_definitions(jobDefinitionName=jd_name, status="ACTIVE") + if not res["jobDefinitions"]: jd_arn = self._register_job_definition(jd_name, container_image, destination_params) else: - jd_arn = res['jobDefinitions'][0]['jobDefinitionArn'] + jd_arn = res["jobDefinitions"][0]["jobDefinitionArn"] LOGGER.debug(f"Found existing job definition: {jd_name}.") return jd_arn def _register_job_definition(self, jd_name, container_image, destination_params): LOGGER.debug(f"Registering a new job definition: {jd_name}.") - platform = destination_params.get('platform') + platform = destination_params.get("platform") # TODO: support multi-node containerProperties = { - 'image': container_image, - 'command': [ - '/bin/sh', + "image": container_image, + "command": [ + "/bin/sh", ], - 'jobRoleArn': destination_params.get('job_role_arn'), - 'executionRoleArn': destination_params.get('execute_role_arn') or destination_params.get('job_role_arn'), - 'volumes': [ + "jobRoleArn": destination_params.get("job_role_arn"), + "executionRoleArn": destination_params.get("execute_role_arn") or destination_params.get("job_role_arn"), + "volumes": [ { - 'name': 'efs_whole', - 'efsVolumeConfiguration': { - 'fileSystemId': destination_params.get('efs_filesystem_id'), - 'rootDirectory': '/', - 'transitEncryption': 'ENABLED', - 'authorizationConfig': { - 'iam': 'ENABLED' - } + "name": "efs_whole", + "efsVolumeConfiguration": { + "fileSystemId": destination_params.get("efs_filesystem_id"), + "rootDirectory": "/", + "transitEncryption": "ENABLED", + "authorizationConfig": {"iam": "ENABLED"} } }, ], - 'mountPoints': [ + "mountPoints": [ { - 'containerPath': destination_params.get('efs_mount_point'), - 'readOnly': False, - 'sourceVolume': 'efs_whole' + "containerPath": destination_params.get("efs_mount_point"), + "readOnly": False, + "sourceVolume": "efs_whole" }, ], - 'resourceRequirements': _add_resource_requirements(destination_params), - 'environment': _add_galaxy_environment_variables( - destination_params.get('vcpu'), - destination_params.get('memory') + "resourceRequirements": _add_resource_requirements(destination_params), + "environment": _add_galaxy_environment_variables( + destination_params.get("vcpu"), destination_params.get("memory") ), - 'user': '%d:%d' % (os.getuid(), os.getgid()), - 'logConfiguration': { - 'logDriver': 'awslogs' - } + "user": "%d:%d" % (os.getuid(), os.getgid()), + "logConfiguration": {"logDriver": "awslogs"} } - if platform == 'FARGATE': + if platform == "FARGATE": containerProperties.update( { - 'networkConfiguration': { - 'assignPublicIp': 'ENABLED' - }, - 'fargatePlatformConfiguration': { - 'platformVersion': destination_params.get('fargate_version') - }, - 'logConfiguration': { - 'logDriver': 'awslogs' - } + "networkConfiguration": {"assignPublicIp": "ENABLED"}, + "fargatePlatformConfiguration": {"platformVersion": destination_params.get("fargate_version")}, + "logConfiguration": {"logDriver": "awslogs"} } ) res = self._batch_client.register_job_definition( jobDefinitionName=jd_name, - type='container', + type="container", platformCapabilities=[platform], - containerProperties=containerProperties + containerProperties=containerProperties, ) - assert res['ResponseMetadata']['HTTPStatusCode'] == 200 - return res['jobDefinitionArn'] + assert res["ResponseMetadata"]["HTTPStatusCode"] == 200 + return res["jobDefinitionArn"] def _submit_job(self, job_def, job_wrapper, destination_params): job_name = self.JOB_NAME_PREFIX + job_wrapper.get_id_tag() @@ -293,18 +280,18 @@ class AWSBatchJobRunner(AsynchronousJobRunner): LOGGER.info(f"Submitting job {job_name} to AWS Batch.") res = self._batch_client.submit_job( jobName=job_name, - jobQueue=destination_params.get('job_queue'), + jobQueue=destination_params.get("job_queue"), jobDefinition=job_def, containerOverrides={ - 'command': [ - '/bin/bash', - f'{command_script_path}', + "command": [ + "/bin/bash", + f"{command_script_path}", ] - } + }, ) - assert res['ResponseMetadata']['HTTPStatusCode'] == 200 - return job_name, res['jobId'] + assert res["ResponseMetadata"]["HTTPStatusCode"] == 200 + return job_name, res["jobId"] @handle_exception_call def stop_job(self, job_wrapper): @@ -374,12 +361,12 @@ class AWSBatchJobRunner(AsynchronousJobRunner): @handle_exception_call def check_watched_items(self): - done = [] # type: ignore + done = [] # type: ignore self.check_watched_items_by_batch(0, len(self.watched), done) self.watched = [x for x in self.watched if x[0] not in done] def check_watched_items_by_batch(self, start: int, end: int, done: List[str]): - jobs = self.watched[start: start + self.MAX_JOBS_PER_QUERY] + jobs = self.watched[start : start + self.MAX_JOBS_PER_QUERY] if not jobs: return @@ -388,20 +375,20 @@ class AWSBatchJobRunner(AsynchronousJobRunner): res = self._batch_client.describe_jobs(jobs=list(jobs_dict.keys())) gotten = [] - for job in res['jobs']: - status = job['status'] - job_id = job['jobId'] + for job in res["jobs"]: + status = job["status"] + job_id = job["jobId"] gotten.append(job_id) job_state = jobs_dict[job_id] - if status == 'SUCCEEDED': + if status == "SUCCEEDED": self._mark_as_successful(job_state) done.append(job_id) - elif status == 'FAILED': - reason = job['statusReason'] + elif status == "FAILED": + reason = job["statusReason"] self._mark_as_failed(job_state, reason) done.append(job_id) - elif status in ('SUBMITTED', 'PENDING', 'RUNNABLE', 'STARTING', 'RUNNING'): + elif status in ("SUBMITTED", "PENDING", "RUNNABLE", "STARTING", "RUNNING"): self._mark_as_active(job_state) # TODO else? @@ -440,58 +427,61 @@ class AWSBatchJobRunner(AsynchronousJobRunner): parsed_params = {} for k, spec in self.DESTINATION_PARAMS_SPEC.items(): value = params.get(k, spec.get("default")) # type: ignore[attr-defined] - if spec.get('required') and not value: # type: ignore[attr-defined] + if spec.get("required") and not value: # type: ignore[attr-defined] check_required.append(k) - mapper = spec.get("map") # type: ignore[attr-defined] + mapper = spec.get("map") # type: ignore[attr-defined] parsed_params[k] = mapper(value) if check_required: - raise AWSBatchRunnerException("AWSBatchJobRunner requires the following params to be provided: %s." - % (', '.join(check_required))) + raise AWSBatchRunnerException( + "AWSBatchJobRunner requires the following params to be provided: %s." % (", ".join(check_required)) + ) # parse Platform - platform = 'EC2' - auto_platform = parsed_params.get('auto_platform') - fargate_version = parsed_params.get('fargate_version') - vcpu = parsed_params.get('vcpu') - memory = parsed_params.get('memory') - gpu = parsed_params.get('gpu') + platform = "EC2" + auto_platform = parsed_params.get("auto_platform") + fargate_version = parsed_params.get("fargate_version") + vcpu = parsed_params.get("vcpu") + memory = parsed_params.get("memory") + gpu = parsed_params.get("gpu") if auto_platform and not fargate_version: raise AWSBatchRunnerException("AWSBatchJobRunner needs 'farget_version' to be set to enable auto platform!") if gpu and (fargate_version or auto_platform): - raise AWSBatchRunnerException("GPU mode is not allowed when 'fargate_version' and/or 'auto_platform' are set!") + raise AWSBatchRunnerException( + "GPU mode is not allowed when 'fargate_version' and/or 'auto_platform' are set!" + ) if fargate_version and not auto_platform: - platform = 'FARGATE' + platform = "FARGATE" if auto_platform: max_vcpu = self.FARGATE_VCPUS[-1] max_memory = self.FARGATE_RESOURCES[max_vcpu][-1] - if vcpu <= max_vcpu and memory <= max_memory: # type: ignore[operator] + if vcpu <= max_vcpu and memory <= max_memory: # type: ignore[operator] new_vcpu, new_memory = None, None for c in self.FARGATE_VCPUS: - if c < vcpu: # type: ignore[operator] + if c < vcpu: # type: ignore[operator] continue for m in self.FARGATE_RESOURCES[c]: - if m >= memory: # type: ignore[operator] + if m >= memory: # type: ignore[operator] new_vcpu = c new_memory = m break if new_memory: - platform = 'FARGATE' - parsed_params['vcpu'] = new_vcpu - parsed_params['memory'] = new_memory + platform = "FARGATE" + parsed_params["vcpu"] = new_vcpu + parsed_params["memory"] = new_memory break # parse JOB QUEUE - job_queues = parsed_params.get('job_queue').split(',') # type: ignore[union-attr] + job_queues = parsed_params.get("job_queue").split(",") # type: ignore[union-attr] if len(job_queues) < 2: raise AWSBatchRunnerException( "AWSBatchJobRunner needs TWO job queues ('Farget Queue, EC2 Qeueue')" " when 'auto_platform' is enabled!" ) - parsed_params['job_queue'] = job_queues[platform == 'EC2'].strip() + parsed_params["job_queue"] = job_queues[platform == "EC2"].strip() - parsed_params['platform'] = platform + parsed_params["platform"] = platform return parsed_params def write_command(self, job_wrapper): From 06caa9ff891421cd364c4f822dae8a465b3148ca Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 10 Mar 2022 18:16:11 -0800 Subject: [PATCH 20/43] Reformat imports --- lib/galaxy/jobs/runners/aws.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 15800354548..7eb0edcca31 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -3,17 +3,20 @@ import hashlib import logging import os import time - from queue import Empty from typing import List -from . import JobState + from galaxy import model from galaxy.job_execution.output_collect import default_exit_code_file from galaxy.jobs.runners import ( + JobState, AsynchronousJobRunner, AsynchronousJobState, ) -from galaxy.util import smart_str, unicodify +from galaxy.util import ( + smart_str, + unicodify, +) BOTO3_IMPORT_MSG = ( From fdb4c01aa12995f272549c130456a2ec134c8c11 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Thu, 10 Mar 2022 10:15:02 -0500 Subject: [PATCH 21/43] Improved API exceptions for serializing views. --- lib/galaxy/jobs/runners/aws.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 7eb0edcca31..6627e4965ce 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -9,16 +9,15 @@ from typing import List from galaxy import model from galaxy.job_execution.output_collect import default_exit_code_file from galaxy.jobs.runners import ( - JobState, AsynchronousJobRunner, AsynchronousJobState, + JobState, ) from galaxy.util import ( smart_str, unicodify, ) - BOTO3_IMPORT_MSG = ( "The Python 'boto3' package is required to use " "this feature, please install it or correct the " From bb37337a3d7ed8934532b5f9e0e1c815ee5bbdd5 Mon Sep 17 00:00:00 2001 From: qiagu Date: Mon, 14 Mar 2022 15:08:05 -0700 Subject: [PATCH 22/43] Add region param to initiate runner client --- lib/galaxy/config/sample/job_conf.xml.sample_advanced | 5 +++-- lib/galaxy/jobs/runners/aws.py | 3 ++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/lib/galaxy/config/sample/job_conf.xml.sample_advanced b/lib/galaxy/config/sample/job_conf.xml.sample_advanced index 58043830f41..876e023ae50 100644 --- a/lib/galaxy/config/sample/job_conf.xml.sample_advanced +++ b/lib/galaxy/config/sample/job_conf.xml.sample_advanced @@ -370,9 +370,10 @@ - + + xxxxxxxxxxxxxxxxxxx + us-west-1 --> + - true + arn_for_Fargate_job_queue, arn_for_EC2_job_queue + arn:aws:iam::xxxxxxxxxxxxxxxxxx 1 2048 fs-xxxxxxxxxxxxxx - /mnt/efs + /mnt/efs/fs1 + 1.4.0 + true true - + arn_for_gpu_job_queue + arn:aws:iam::xxxxxxxxxxxxxxxxxx 4 20000 1 fs-xxxxxxxxxxxxxx - /mnt/efs + /mnt/efs/fs1 + + + + + + + true + + arn_for_Fargate_job_queue, arn_for_EC2_job_queue + + arn:aws:iam::xxxxxxxxxxxxxxxxxx + 1 + 2048 + + fs-xxxxxxxxxxxxxx + /mnt/efs/fs1 + + 1.4.0 + + true + + + true + + arn_for_gpu_job_queue + + arn:aws:iam::xxxxxxxxxxxxxxxxxx + 4 + 20000 + + 1 + fs-xxxxxxxxxxxxxx + /mnt/efs/fs1 + + + +``` + ## Submitting Jobs as the Real User Galaxy runs as a process on your server as whatever user starts the server - usually an account created for the purpose of running Galaxy. Jobs will be submitted to your cluster(s) as this user. In environments where users in Galaxy are guaranteed to be users on the underlying system (i.e. Galaxy is configured to use external authentication), it may be desirable to submit jobs to the cluster as the user logged in to Galaxy rather than Galaxy's system user. From c4c7fbc5c2b625cbf5e90f17df0f8b798314150a Mon Sep 17 00:00:00 2001 From: qiagu Date: Sun, 5 Jun 2022 16:22:49 -0700 Subject: [PATCH 34/43] Add aws batch retry strategy --- .../sample/job_conf.xml.sample_advanced | 7 +++ lib/galaxy/jobs/runners/aws.py | 47 ++++++++++++++++++- 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/lib/galaxy/config/sample/job_conf.xml.sample_advanced b/lib/galaxy/config/sample/job_conf.xml.sample_advanced index cfefffe1200..fe5e6aaee06 100644 --- a/lib/galaxy/config/sample/job_conf.xml.sample_advanced +++ b/lib/galaxy/config/sample/job_conf.xml.sample_advanced @@ -1151,6 +1151,13 @@ 1.4.0 true + + true diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index deef3beb0f5..1bdc122b8f1 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -162,7 +162,23 @@ class AWSBatchJobRunner(AsynchronousJobRunner): "privileged": { "default": False, "map": lambda x: str(x).lower() == "true", - } + }, + "retry_attempts": { + "default": 1, + "map": int, + }, + "retry_on_exit_statusReason": { + "default": "", + "map": str, + }, + "retry_on_exit_reason": { + "default": "", + "map": str, + }, + "retry_on_exit_exitCode": { + "default": "", + "map": str, + }, } FARGATE_VCPUS = [0.25, 0.5, 1, 2, 4] @@ -274,6 +290,30 @@ class AWSBatchJobRunner(AsynchronousJobRunner): ) return volumes, mount_points + def _get_retry_strategy(self, destination_params): + """ Make a simple one-condition retry strategy + """ + # TODO make multiple-condition retry strategies + attemps = destination_params.get("retry_attempts") + status_reason = destination_params.get("retry_on_exit_statusReason") + reason = destination_params.get("retry_on_exit_reason") + exit_code = destination_params.get("retry_on_exit_exitCode") + + if attemps <= 1: + return + + strategy = { + "attempts": attemps, + "evaluateOnExit": [{"action": "RETRY"}], + } + if status_reason: + strategy["evaluateOnExit"][0]["onStatusReason"] = status_reason + if reason: + strategy["evaluateOnExit"][0]["onReason"] = reason + if exit_code: + strategy["evaluateOnExit"][0]["onExitCode"] = exit_code + return strategy + def _register_job_definition(self, jd_name, container_image, destination_params): log.debug(f"Registering a new job definition: {jd_name}.") platform = destination_params.get("platform") @@ -305,12 +345,17 @@ class AWSBatchJobRunner(AsynchronousJobRunner): "logConfiguration": {"logDriver": "awslogs"}, } ) + other_kwargs = {} + retry_strategy = self._get_retry_strategy(destination_params) + if retry_strategy: + other_kwargs["retryStrategy"] = retry_strategy res = self._batch_client.register_job_definition( jobDefinitionName=jd_name, type="container", platformCapabilities=[platform], containerProperties=containerProperties, + **other_kwargs, ) assert res["ResponseMetadata"]["HTTPStatusCode"] == 200 From 049a9ee1d877c80cd4aef69a6e4214ff13e4a748 Mon Sep 17 00:00:00 2001 From: Qiang Gu <36086881+qiagu@users.noreply.github.com> Date: Mon, 6 Jun 2022 09:46:32 -0700 Subject: [PATCH 35/43] Update doc/source/admin/cluster.md Co-authored-by: Nuwan Goonasekera <2070605+nuwang@users.noreply.github.com> --- doc/source/admin/cluster.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/source/admin/cluster.md b/doc/source/admin/cluster.md index 9b3a0829bb8..8a9a14de329 100644 --- a/doc/source/admin/cluster.md +++ b/doc/source/admin/cluster.md @@ -310,7 +310,7 @@ Most options available to `qsub(1b)` and `pbs_submit(3b)` are supported. Except ## AWS Batch -Runs jobs via the [AWS Batch](https://aws.amazon.com/batch/). Built on top of AWS Elastic Container Service (ECS), AWS Batch enables users to run hundreds of thousands of jobs with simple configuration. +Runs jobs via [AWS Batch](https://aws.amazon.com/batch/). Built on top of AWS Elastic Container Service (ECS), AWS Batch enables users to run hundreds of thousands of jobs with little configuration. #### Dependencies From 036e3a383142b17b0f80abc4ea32d11434ab7161 Mon Sep 17 00:00:00 2001 From: Qiang Gu <36086881+qiagu@users.noreply.github.com> Date: Mon, 6 Jun 2022 09:47:20 -0700 Subject: [PATCH 36/43] Update lib/galaxy/config/sample/job_conf.xml.sample_advanced Co-authored-by: Nuwan Goonasekera <2070605+nuwang@users.noreply.github.com> --- lib/galaxy/config/sample/job_conf.xml.sample_advanced | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/galaxy/config/sample/job_conf.xml.sample_advanced b/lib/galaxy/config/sample/job_conf.xml.sample_advanced index fe5e6aaee06..e6acf70e9b9 100644 --- a/lib/galaxy/config/sample/job_conf.xml.sample_advanced +++ b/lib/galaxy/config/sample/job_conf.xml.sample_advanced @@ -1141,7 +1141,7 @@ true arn_for_Fargate_job_queue, arn_for_EC2_job_queue - + arn:aws:iam::xxxxxxxxxxxxxxxxxx 1 2048 From 92ec7c60b145a825368330ae031ade8bcd11b2de Mon Sep 17 00:00:00 2001 From: Qiang Gu <36086881+qiagu@users.noreply.github.com> Date: Mon, 6 Jun 2022 12:13:35 -0700 Subject: [PATCH 37/43] Update doc/source/admin/cluster.md Co-authored-by: Nuwan Goonasekera <2070605+nuwang@users.noreply.github.com> --- doc/source/admin/cluster.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/doc/source/admin/cluster.md b/doc/source/admin/cluster.md index 8a9a14de329..55900e12841 100644 --- a/doc/source/admin/cluster.md +++ b/doc/source/admin/cluster.md @@ -315,12 +315,12 @@ Runs jobs via [AWS Batch](https://aws.amazon.com/batch/). Built on top of AWS El #### Dependencies AWS Batch job runner requirs AWS Elastic File System (EFS) being mounted as a shared file system that enables Galaxy and job containers to read and write files. In the best pratice, Galaxy is installed an AWS EC2 instance and an EFS is mounted to the EC2 as a local drive. Job-related paths, such as objects, jobs_directory, tool_directory and so on, need to be placed on the EFS drive. -In addition, Galaxy admin needs to configure Batch compute environment, Batch job queue and proper AWS IAM roles, and provision them as destination parameters. -AWS Batch job runner requires [boto3](https://pypi.org/project/boto3/) installed in Galaxy environment. +In addition, a Galaxy admin needs to configure a Batch compute environment, a Batch job queue and proper AWS IAM roles, and specify them as destination parameters. +AWS Batch job runner requires [boto3](https://pypi.org/project/boto3/) to be installed in Galaxy's environment. #### Parameters and Configuration -AWS Batch job runner sends jobs to Batch compute environment that is composed of either Fargate or EC2. While Fargate provides a series of lightweigt compute resources (up to 4 vcpu and 30 GB memeory), the EC2 offers more abroad choices. With `auto_platform` enabled, this runner supports mapping to the best fit type of resources based on the provisioned `vcpu` and `memory`, i.e., Fargate is preferred over EC2 when `vcpu` and `memory` don't go beyond the limits (4 and 30 gb, respectively). If the power of `GPU` is needed for a destination, a job queue built on top of GPU-enabled compute environment must be provisoned. +AWS Batch job runner sends jobs to Batch compute environment that is composed of either Fargate or EC2. While Fargate provides a series of lightweight compute resources (up to 4 vcpu and 30 GB memory), EC2 offers broader choices. With `auto_platform` enabled, this runner supports mapping to the best fit type of resources based on the requested `vcpu` and `memory`, i.e., Fargate is preferred over EC2 when `vcpu` and `memory` are below the limits (4 and 30 gb, respectively). If `GPU` computing is needed for a destination, a job queue built on top of a GPU-enabled compute environment must be provisioned. ```xml From 419b6835d55978af7c85380acf58ffa5530cac4f Mon Sep 17 00:00:00 2001 From: qiagu Date: Tue, 14 Jun 2022 11:17:52 -0700 Subject: [PATCH 38/43] Reword dependencies in Batch job runner doc --- doc/source/admin/cluster.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/doc/source/admin/cluster.md b/doc/source/admin/cluster.md index 55900e12841..1f81ef6fcd4 100644 --- a/doc/source/admin/cluster.md +++ b/doc/source/admin/cluster.md @@ -314,8 +314,7 @@ Runs jobs via [AWS Batch](https://aws.amazon.com/batch/). Built on top of AWS El #### Dependencies -AWS Batch job runner requirs AWS Elastic File System (EFS) being mounted as a shared file system that enables Galaxy and job containers to read and write files. In the best pratice, Galaxy is installed an AWS EC2 instance and an EFS is mounted to the EC2 as a local drive. Job-related paths, such as objects, jobs_directory, tool_directory and so on, need to be placed on the EFS drive. -In addition, a Galaxy admin needs to configure a Batch compute environment, a Batch job queue and proper AWS IAM roles, and specify them as destination parameters. +The AWS Batch job runner requires AWS Elastic File System (EFS) to be mounted as a shared file system that enables both Galaxy and job containers to read and write files. In a typical use case, Galaxy is installed on an AWS EC2 instance where an EFS drive is mounted, and all job-related paths, such as objects, jobs_directory, tool_directory and so on, are placed on the EFS drive. Galaxy admins configure Batch compute environments, Batch job queues and proper AWS IAM roles, and specify them as destination parameters. AWS Batch job runner requires [boto3](https://pypi.org/project/boto3/) to be installed in Galaxy's environment. #### Parameters and Configuration From 728d089c18ca2587a55edeeb9e4aef4cc842da36 Mon Sep 17 00:00:00 2001 From: qiagu Date: Tue, 21 Jun 2022 11:22:21 -0700 Subject: [PATCH 39/43] Allow retry action to be exit --- lib/galaxy/config/sample/job_conf.xml.sample_advanced | 1 + lib/galaxy/jobs/runners/aws.py | 9 +++++++-- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/lib/galaxy/config/sample/job_conf.xml.sample_advanced b/lib/galaxy/config/sample/job_conf.xml.sample_advanced index e6acf70e9b9..9ac22fcd378 100644 --- a/lib/galaxy/config/sample/job_conf.xml.sample_advanced +++ b/lib/galaxy/config/sample/job_conf.xml.sample_advanced @@ -1157,6 +1157,7 @@ ... ... ... + exit --> diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 1bdc122b8f1..3defa7f3f9d 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -179,6 +179,10 @@ class AWSBatchJobRunner(AsynchronousJobRunner): "default": "", "map": str, }, + "retry_on_exit_action": { + "default": "RETRY", + "map": str, + }, } FARGATE_VCPUS = [0.25, 0.5, 1, 2, 4] @@ -293,18 +297,19 @@ class AWSBatchJobRunner(AsynchronousJobRunner): def _get_retry_strategy(self, destination_params): """ Make a simple one-condition retry strategy """ - # TODO make multiple-condition retry strategies + # TODO make multi-condition retry strategies attemps = destination_params.get("retry_attempts") status_reason = destination_params.get("retry_on_exit_statusReason") reason = destination_params.get("retry_on_exit_reason") exit_code = destination_params.get("retry_on_exit_exitCode") + action = destination_params.get("retry_on_exit_action") if attemps <= 1: return strategy = { "attempts": attemps, - "evaluateOnExit": [{"action": "RETRY"}], + "evaluateOnExit": [{"action": action}], } if status_reason: strategy["evaluateOnExit"][0]["onStatusReason"] = status_reason From de08e39dc34b523f84b8904d32f8344dc9b7acaa Mon Sep 17 00:00:00 2001 From: qiagu Date: Sat, 23 Jul 2022 12:05:01 -0700 Subject: [PATCH 40/43] Improve auto_platform resources mapping --- lib/galaxy/jobs/runners/aws.py | 22 ++++++++++------------ 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 3defa7f3f9d..fcb7c42f1f3 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -1,5 +1,6 @@ """ Galaxy job runners to use Amazon AWS native compute resources, such as AWS Batch. """ +import bisect import hashlib import json import logging @@ -548,20 +549,17 @@ class AWSBatchJobRunner(AsynchronousJobRunner): max_vcpu = self.FARGATE_VCPUS[-1] max_memory = self.FARGATE_RESOURCES[max_vcpu][-1] if vcpu <= max_vcpu and memory <= max_memory: # type: ignore[operator] - new_vcpu, new_memory = None, None - for c in self.FARGATE_VCPUS: - if c < vcpu: # type: ignore[operator] - continue - for m in self.FARGATE_RESOURCES[c]: - if m >= memory: # type: ignore[operator] - new_vcpu = c - new_memory = m - break - if new_memory: + c_ix = bisect.bisect_left(self.FARGATE_VCPUS, vcpu) # type: ignore[type-var] + length = len(self.FARGATE_VCPUS) + while c_ix < length: + c = self.FARGATE_VCPUS[c_ix] + m_ix = bisect.bisect_left(self.FARGATE_RESOURCES[c], memory) # type: ignore[type-var] + if m_ix < len(self.FARGATE_RESOURCES[c]): platform = "FARGATE" - parsed_params["vcpu"] = new_vcpu - parsed_params["memory"] = new_memory + parsed_params["vcpu"] = c + parsed_params["memory"] = self.FARGATE_RESOURCES[c][m_ix] break + c_ix += 1 # parse JOB QUEUE job_queues = parsed_params.get("job_queue").split(",") # type: ignore[union-attr] if len(job_queues) < 2: From 56856d0e6297d554b5c601fc712f105d116528b6 Mon Sep 17 00:00:00 2001 From: qiagu Date: Sun, 24 Jul 2022 14:00:35 -0700 Subject: [PATCH 41/43] Don't use farget_vcpus as a property --- lib/galaxy/jobs/runners/aws.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index fcb7c42f1f3..a3d26e6a1e9 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -186,8 +186,6 @@ class AWSBatchJobRunner(AsynchronousJobRunner): }, } - FARGATE_VCPUS = [0.25, 0.5, 1, 2, 4] - # fmt: off FARGATE_RESOURCES = { 0.25: [512, 1024, 2048], @@ -546,13 +544,14 @@ class AWSBatchJobRunner(AsynchronousJobRunner): if fargate_version and not auto_platform: platform = "FARGATE" if auto_platform: - max_vcpu = self.FARGATE_VCPUS[-1] + fargate_vcpus = list(self.FARGATE_RESOURCES.keys()) + max_vcpu = fargate_vcpus[-1] max_memory = self.FARGATE_RESOURCES[max_vcpu][-1] if vcpu <= max_vcpu and memory <= max_memory: # type: ignore[operator] - c_ix = bisect.bisect_left(self.FARGATE_VCPUS, vcpu) # type: ignore[type-var] - length = len(self.FARGATE_VCPUS) + c_ix = bisect.bisect_left(fargate_vcpus, vcpu) # type: ignore[type-var] + length = len(fargate_vcpus) while c_ix < length: - c = self.FARGATE_VCPUS[c_ix] + c = fargate_vcpus[c_ix] m_ix = bisect.bisect_left(self.FARGATE_RESOURCES[c], memory) # type: ignore[type-var] if m_ix < len(self.FARGATE_RESOURCES[c]): platform = "FARGATE" From 5bc8827510df04dec70af175b3beceeef8365451 Mon Sep 17 00:00:00 2001 From: qiagu Date: Thu, 4 Aug 2022 15:01:40 -0700 Subject: [PATCH 42/43] Improve batch status checker --- lib/galaxy/jobs/runners/aws.py | 25 +++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index a3d26e6a1e9..51729eac526 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -8,7 +8,7 @@ import os import re import time from queue import Empty -from typing import List +from typing import Set from galaxy import model from galaxy.job_execution.output_collect import default_exit_code_file @@ -451,42 +451,43 @@ class AWSBatchJobRunner(AsynchronousJobRunner): time.sleep(max(self.app.config.job_runner_monitor_sleep, self.MIN_QUERY_INTERVAL)) def check_watched_items(self): - done = [] # type: ignore + done: Set[str] = set() self.check_watched_items_by_batch(0, len(self.watched), done) self.watched = [x for x in self.watched if x[0] not in done] - def check_watched_items_by_batch(self, start: int, end: int, done: List[str]): + def check_watched_items_by_batch(self, start: int, end: int, done: Set[str]): jobs = self.watched[start : start + self.MAX_JOBS_PER_QUERY] if not jobs: return jobs_dict = dict(jobs) + resp = self._batch_client.describe_jobs(jobs=list(jobs_dict.keys())) - res = self._batch_client.describe_jobs(jobs=list(jobs_dict.keys())) - - gotten = [] - for job in res["jobs"]: + gotten = set() + for job in resp["jobs"]: status = job["status"] job_id = job["jobId"] - gotten.append(job_id) + gotten.add(job_id) job_state = jobs_dict[job_id] if status == "SUCCEEDED": self._mark_as_successful(job_state) - done.append(job_id) + done.add(job_id) elif status == "FAILED": reason = job["statusReason"] self._mark_as_failed(job_state, reason) - done.append(job_id) + done.add(job_id) elif status in ("SUBMITTED", "PENDING", "RUNNABLE", "STARTING", "RUNNING"): self._mark_as_active(job_state) # TODO else? - for job_id in set(jobs_dict.keys()) - set(gotten): + for job_id in jobs_dict: + if job_id in gotten: + continue job_state = jobs_dict[job_id] reason = f"The track of Job {job_state} was lost for unknown reason!" self._mark_as_failed(job_state, reason) - done.append(job_id) + done.add(job_id) self.check_watched_items_by_batch(start + self.MAX_JOBS_PER_QUERY, end, done) From 5bd6c27fa992b568081a2e6bb7254a71fbd0f53c Mon Sep 17 00:00:00 2001 From: qiagu Date: Sun, 14 Aug 2022 15:07:28 -0700 Subject: [PATCH 43/43] Extend Galaxy queued job status to reflect AWS status --- lib/galaxy/jobs/runners/aws.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py index 51729eac526..8040200d2ca 100644 --- a/lib/galaxy/jobs/runners/aws.py +++ b/lib/galaxy/jobs/runners/aws.py @@ -477,8 +477,9 @@ class AWSBatchJobRunner(AsynchronousJobRunner): reason = job["statusReason"] self._mark_as_failed(job_state, reason) done.add(job_id) - elif status in ("SUBMITTED", "PENDING", "RUNNABLE", "STARTING", "RUNNING"): + elif status in ("STARTING", "RUNNING"): self._mark_as_active(job_state) + # remain queued for "SUBMITTED", "PENDING" and "RUNNABLE" # TODO else? for job_id in jobs_dict: