Remove set permissions

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": {

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()
 

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
 
 

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)

ridgeback/__init__.py

@@ -1 +1 @@
-__version__ = "2.1.4"
+__version__ = "2.1.6"

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)
 

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",
             ]

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")