diff --git a/orchestrator/celery.py b/orchestrator/celery.py index 18691d45..8e30260d 100644 --- a/orchestrator/celery.py +++ b/orchestrator/celery.py @@ -21,14 +21,13 @@ app.conf.task_routes = { "orchestrator.tasks.cleanup_folders": {"queue": settings.RIDGEBACK_ACTION_QUEUE}, "orchestrator.tasks.command_processor": {"queue": settings.RIDGEBACK_COMMAND_QUEUE}, - "orchestrator.tasks.set_permissions_job": {"queue": settings.RIDGEBACK_SET_PERMISSIONS_QUEUE}, "orchestrator.tasks.check_leader_not_running": {"queue": settings.RIDGEBACK_CHECK_STATUS_QUEUE}, } app.conf.beat_schedule = { "process_jobs": { "task": "orchestrator.tasks.process_jobs", - "schedule": 60.0, + "schedule": settings.RIDGEBACK_CHECK_JOBS_INTERVAL, "options": {"queue": settings.RIDGEBACK_SUBMIT_JOB_QUEUE}, }, "cleanup_completed_jobs": { diff --git a/orchestrator/models.py b/orchestrator/models.py index 6f2f57c8..5ee8787e 100755 --- a/orchestrator/models.py +++ b/orchestrator/models.py @@ -39,7 +39,7 @@ class Status(IntEnum): SUBMITTED = 3 PENDING = 4 RUNNING = 5 - SET_PERMISSIONS = 6 + SET_PERMISSIONS = 6 # DEPRECATED STATE COMPLETED = 7 FAILED = 8 TERMINATED = 9 @@ -247,7 +247,7 @@ def update_status(self, lsf_status): def pipeline_completed(self, outputs): self.track_cache = None self.outputs = outputs - self.status = Status.SET_PERMISSIONS + self.status = Status.COMPLETED self.finished = now() self.save() diff --git a/orchestrator/tasks.py b/orchestrator/tasks.py index 271bd251..4153647a 100644 --- a/orchestrator/tasks.py +++ b/orchestrator/tasks.py @@ -3,8 +3,6 @@ import shutil import logging import tempfile -from pathlib import Path -from getpass import getuser from datetime import timedelta from celery import shared_task from django.conf import settings @@ -183,16 +181,6 @@ def reset_job_to_created(job_id): job.save() -@shared_task(bind=True) -def set_permissions_job(self, job_id): - job = Job.objects.get(id=job_id) - try: - set_permission(job) - except Exception as e: - logger.error(f"Failed to set permissions for job:{job_id}. {str(e)}") - job.complete() - - def prepare_job(job): if Status(job.status).transition(Status.PREPARED): logger.info(f"Preparing job {str(job.id)} for execution") @@ -265,7 +253,6 @@ def submit_job_to_batch_system(job, retries=0): def _pipeline_completed(job, outputs): job.pipeline_completed(outputs) - # set_permissions_job.delay(str(job.id)) def _fail(job, error_message=""): @@ -329,7 +316,9 @@ def check_job_status(job): job.update_status(batch_system_status) if batch_system_status in (Status.RUNNING,): - command_processor.delay(Command(CommandType.CHECK_HANGING, str(job.id)).to_dict()) + # TODO: Fix performance and errors in CHECK_HANGING and CHECK_COMMAND_LINE_STATUS commands + pass + # command_processor.delay(Command(CommandType.CHECK_HANGING, str(job.id)).to_dict()) # command_processor.delay(Command(CommandType.CHECK_COMMAND_LINE_STATUS, str(job.id)).to_dict()) elif batch_system_status in (Status.COMPLETED,): @@ -481,46 +470,6 @@ def terminate_job(job): job.terminate() -@userswitch -def set_permission(job): - failed_to_set = None - dirs = job.root_dir.replace(job.base_dir, "").split("/") - permission_str = job.root_permission - permissions_dir = job.base_dir - for d in dirs: - failed_to_set = False - permissions_dir = "/".join([permissions_dir, d]).replace("//", "/") - try: - permission_octal = int(permission_str, 8) - except Exception: - raise TypeError("Could not convert %s to permission octal" % str(permission_str)) - try: - if Path(permissions_dir).owner() == getuser(): - os.chmod(permissions_dir, permission_octal) - else: - logger.debug(f"Skipping permission change for {permissions_dir} as it is not owned by {getuser()}") - for root, dirs, files in os.walk(permissions_dir): - for single_dir in dirs: - if oct(os.lstat(os.path.join(root, single_dir)).st_mode)[-3:] != permission_octal: - logger.debug(f"Setting permissions for {os.path.join(root, single_dir)}") - path = os.path.join(root, single_dir) - os.chmod(path, permission_octal) - for single_file in files: - if oct(os.lstat(os.path.join(root, single_file)).st_mode)[-3:] != permission_octal: - path = os.path.join(root, single_file) - logger.debug(f"Setting permissions for {path}") - os.chmod(path, permission_octal) - except Exception: - logger.exception(f"Failed to set permissions for directory {permissions_dir}") - failed_to_set = True - continue - else: - logger.debug(f"Permissions set for directory {permissions_dir}") - break - if failed_to_set: - raise RuntimeError("Failed to change permission of directory %s" % permissions_dir) - - # Cleaning jobs diff --git a/orchestrator/tests/test_tasks.py b/orchestrator/tests/test_tasks.py index 226e863a..2f5e2052 100644 --- a/orchestrator/tests/test_tasks.py +++ b/orchestrator/tests/test_tasks.py @@ -1,8 +1,6 @@ import uuid from mock import patch, call from django.test import TestCase -import tempfile -import os from orchestrator.commands import CommandType, Command from orchestrator.models import CommandLineToolJob from orchestrator.models import Job, Status, PipelineType @@ -13,7 +11,6 @@ process_jobs, cleanup_folders, get_job_info_path, - set_permission, ) @@ -206,8 +203,7 @@ def test_pending_to_running(self, status, command_processor): @patch("orchestrator.tasks.command_processor.delay") @patch("submitter.toil_submitter.ToilJobSubmitter.get_outputs") @patch("batch_systems.lsf_client.lsf_client.LSFClient.status") - @patch("orchestrator.tasks.set_permissions_job.delay") - def test_running_to_pipeline_completed(self, permission, status, get_outputs, command_processor): + def test_running_to_pipeline_completed(self, status, get_outputs, command_processor): job = Job.objects.create( type=PipelineType.CWL, app={ @@ -221,14 +217,13 @@ def test_running_to_pipeline_completed(self, permission, status, get_outputs, co status=Status.RUNNING, metadata={"pipeline_name": "NA"}, ) - permission.return_value = None status.return_value = Status.COMPLETED, "" outputs = {"output": "test_value"} get_outputs.return_value = outputs, None command_processor.return_value = None check_job_status(job) job.refresh_from_db() - self.assertEqual(job.status, Status.SET_PERMISSIONS) + self.assertEqual(job.status, Status.COMPLETED) self.assertEqual(job.outputs, outputs) @patch("orchestrator.tasks.command_processor.delay") @@ -561,68 +556,3 @@ def test_get_job_info_path(self): with self.settings(PIPELINE_CONFIG=PIPELINE_CONFIG): res = get_job_info_path(str(job.id)) self.assertEqual(res, f"{str(job.working_dir)}/.run.info") - - def test_permission(self): - with tempfile.TemporaryDirectory() as temp_path: - expected_permission = "750" - job_completed = Job.objects.create( - type=PipelineType.CWL, - app={ - "github": { - "version": "1.0.0", - "entrypoint": "test.cwl", - "repository": "", - } - }, - root_dir=temp_path, - base_dir="/".join(temp_path.split("/")[:-1]) + "/", - root_permission=expected_permission, - external_id="ext_id", - status=Status.COMPLETED, - metadata={"pipeline_name": "NA"}, - ) - set_permission(job_completed) - current_permission = oct(os.stat(temp_path).st_mode)[-3:] - self.assertEqual(current_permission, expected_permission) - - def test_permission_wrong_permission(self): - with self.assertRaises(TypeError): - with tempfile.TemporaryDirectory() as temp_path: - expected_permission = "auk" - job_completed = Job.objects.create( - type=PipelineType.CWL, - app={ - "github": { - "version": "1.0.0", - "entrypoint": "test.cwl", - "repository": "", - } - }, - root_dir=temp_path, - base_dir="/".join(temp_path.split("/")[:-1]) + "/", - root_permission=expected_permission, - external_id="ext_id", - status=Status.COMPLETED, - metadata={"pipeline_name": "NA"}, - ) - set_permission(job_completed) - - def test_permission_wrong_path(self): - with self.assertRaises(RuntimeError): - expected_permission = "750" - job_completed = Job.objects.create( - type=PipelineType.CWL, - app={ - "github": { - "version": "1.0.0", - "entrypoint": "test.cwl", - "repository": "", - } - }, - root_dir="/awk", - root_permission=expected_permission, - external_id="ext_id", - status=Status.COMPLETED, - metadata={"pipeline_name": "NA"}, - ) - set_permission(job_completed) diff --git a/ridgeback/__init__.py b/ridgeback/__init__.py index df4be5e0..edc60b35 100644 --- a/ridgeback/__init__.py +++ b/ridgeback/__init__.py @@ -1 +1 @@ -__version__ = "2.1.4" +__version__ = "2.1.6" diff --git a/ridgeback/settings.py b/ridgeback/settings.py index 29d88539..f428e8d9 100644 --- a/ridgeback/settings.py +++ b/ridgeback/settings.py @@ -20,7 +20,7 @@ SHORT_JOB_MAX_DURATION = int(os.environ.get("SHORT_JOB_MAX_DURATION", 4321)) MEDIUM_JOB_QUEUE = int(os.environ.get("MEDIUM_JOB_QUEUE_SIZE", 100)) MEDIUM_JOB_MAX_DURATION = int(os.environ.get("MEDIUM_JOB_MAX_DURATION", 7201)) -LONG_JOB_QUEUE = int(os.environ.get("LONG_JOB_QUEUE_SIZE", 150)) +LONG_JOB_QUEUE = int(os.environ.get("LONG_JOB_QUEUE_SIZE", 100)) # Quick-start development settings - unsuitable for production # See https://docs.djangoproject.com/en/2.2/howto/deployment/checklist/ @@ -203,7 +203,7 @@ RIDGEBACK_SUBMIT_JOB_LSF_QUEUE = os.environ.get("RIDGEBACK_SUBMIT_JOB_LSF_QUEUE", "ridgeback_submit_job_lsf_queue") RIDGEBACK_CLEANUP_QUEUE = os.environ.get("RIDGEBACK_CLEANUP_QUEUE", "ridgeback_cleanup_queue") RIDGEBACK_COMMAND_QUEUE = os.environ.get("RIDGEBACK_COMMAND_QUEUE", "ridgeback_command_queue") -RIDGEBACK_SET_PERMISSIONS_QUEUE = os.environ.get("RIDGEBACK_SET_PERMISSIONS_QUEUE", "ridgeback_set_permissions") +RIDGEBACK_CHECK_JOBS_INTERVAL = int(os.environ.get("RIDGEBACK_CHECK_JOBS_INTERVAL", 180)) CELERY_ACCEPT_CONTENT = ["json"] CELERY_TASK_SERIALIZER = "json" @@ -213,6 +213,12 @@ LOG_PATH = os.environ.get("RIDGEBACK_LOG_PATH", "ridgeback-server.log") +if ENVIRONMENT == "prod": + handlers = ["file"] +else: + handlers = ["file", "console"] + + LOGGING = { "version": 1, "disable_existing_loggers": False, @@ -227,9 +233,9 @@ }, }, "loggers": { - "django_auth_ldap": {"level": "DEBUG", "handlers": ["console"]}, + "django_auth_ldap": {"level": "DEBUG", "handlers": handlers}, "django": { - "handlers": ["file", "console"], + "handlers": handlers, "level": "INFO", "propagate": True, }, @@ -297,6 +303,7 @@ TOIL_STATE_POLLING_WAIT = os.environ.get("TOIL_STATE_POLLING_WAIT", 60) TOIL_MAX_CORES = os.environ.get("RIDGEBACK_TOIL_MAX_CORES", "40") TOIL_DEFAULT_MEMORY = os.environ.get("RIDGEBACK_TOIL_DEFAULT_MEMORY", "8G") +TOIL_JOB_STORE_TIMEOUT = os.environ.get("RIDGEBACK_TOIL_JOB_STORE_TIMEOUT", "86400") SINGLE_MACHINE_CORES = os.environ.get("RIDGEBACK_SINGLE_MACHINE_CORES", 16) SINGLE_MACHINE_MEMORY = os.environ.get("RIDGEBACK_SINGLE_MACHINE_MEMORY", 25) diff --git a/submitter/toil_submitter/toil_jobsubmitter.py b/submitter/toil_submitter/toil_jobsubmitter.py index b64f7eda..afa88d8f 100644 --- a/submitter/toil_submitter/toil_jobsubmitter.py +++ b/submitter/toil_submitter/toil_jobsubmitter.py @@ -263,7 +263,7 @@ def _command_line(self): "--maxCores", settings.TOIL_MAX_CORES, "--jobStoreTimeout", - "600", + settings.TOIL_JOB_STORE_TIMEOUT, "--maxMemory", "256G", "--not-strict", @@ -279,7 +279,7 @@ def _command_line(self): "--outdir", self.job_outputs_dir, "--maxLocalJobs", - "500", + "300", "--no-prepull", "--reference-inputs", ] @@ -314,7 +314,7 @@ def _command_line(self): "--maxCores", settings.TOIL_MAX_CORES, "--jobStoreTimeout", - "600", + settings.TOIL_JOB_STORE_TIMEOUT, "--maxMemory", "256G", "--not-strict", @@ -330,7 +330,7 @@ def _command_line(self): "--outdir", self.job_outputs_dir, "--maxLocalJobs", - "500", + "300", "--no-prepull", "--reference-inputs", ] diff --git a/tests/test_tasks.py b/tests/test_tasks.py index c953261c..89970cb6 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -343,38 +343,34 @@ def test_job_args_all_options_slurm(self): @patch("orchestrator.tasks.get_job_info_path") @patch("batch_systems.lsf_client.lsf_client.LSFClient.status") @patch("submitter.toil_submitter.ToilJobSubmitter.get_outputs") - @patch("orchestrator.tasks.set_permissions_job.delay") - def test_complete_lsf(self, permission, get_outputs, status, get_job_info_path, command_processor): + def test_complete_lsf(self, get_outputs, status, get_job_info_path, command_processor): with override_settings(BATCH_SYSTEM="LSF"): self.current_job.status = Status.PENDING self.current_job.save() - permission.return_value = None command_processor.return_value = True get_outputs.return_value = {"outputs": True}, None get_job_info_path.return_value = "sample/job/path" status.return_value = Status.COMPLETED, None check_job_status(self.current_job) self.current_job.refresh_from_db() - self.assertEqual(self.current_job.status, Status.SET_PERMISSIONS) + self.assertEqual(self.current_job.status, Status.COMPLETED) self.assertNotEqual(self.current_job.finished, None) @patch("orchestrator.tasks.command_processor.delay") @patch("orchestrator.tasks.get_job_info_path") @patch("batch_systems.slurm_client.slurm_client.SLURMClient.status") @patch("submitter.toil_submitter.ToilJobSubmitter.get_outputs") - @patch("orchestrator.tasks.set_permissions_job.delay") - def test_complete_slurm(self, permission, get_outputs, status, get_job_info_path, command_processor): + def test_complete_slurm(self, get_outputs, status, get_job_info_path, command_processor): with override_settings(BATCH_SYSTEM="SLURM"): self.current_job.status = Status.PENDING self.current_job.save() - permission.return_value = None command_processor.return_value = True get_outputs.return_value = {"outputs": True}, None get_job_info_path.return_value = "sample/job/path" status.return_value = Status.COMPLETED, None check_job_status(self.current_job) self.current_job.refresh_from_db() - self.assertEqual(self.current_job.status, Status.SET_PERMISSIONS) + self.assertEqual(self.current_job.status, Status.COMPLETED) self.assertNotEqual(self.current_job.finished, None) @patch("orchestrator.tasks.command_processor.delay")