Merge pull request #194 from hubmapconsortium/phillips/custom_parameters

Custom parameter support

src/example_config.json

@@ -29,7 +29,12 @@
       "storage": "main_storage",
       "user_authentication": "main_auth",
       "passthrough_domain": "127.0.0.1:8000",
-      "connection_details": {}
+      "connection_details": {},
+      "parameter_mapping": {
+            "num_cpus": "cpus_per_task",
+            "memory_mb": "memory_per_node",
+            "time_limit_min": "time_limit"
+       }
     }
   },
   "available_job_types": {
@@ -56,5 +61,53 @@
         }
       }
     }
-  }
+  },
+  "parameters": [
+    {
+      "display_name":  "Number of CPUs",
+      "description": "",
+      "variable_name": "num_cpus",
+      "default_value": 1,
+      "validation": {
+        "type": "int",
+        "min": 1,
+        "max": 4,
+        "required": false
+      }
+    },
+    {
+      "display_name": "Memory (MB)",
+      "description": "",
+      "variable_name": "memory_mb",
+      "default_value": 1024,
+      "validation": {
+        "type": "int",
+        "min": 1024,
+        "max": 8192,
+        "required": false
+      }
+    },
+    {
+      "display_name": "Time Limit (minutes)",
+      "description": "",
+      "variable_name": "time_limit_min",
+      "default_value": 180,
+      "validation": {
+        "type": "int",
+        "min": 1,
+        "max": 480,
+        "required": false
+      }
+    },
+    {
+      "display_name": "GPU Enabled",
+      "description": "",
+      "variable_name": "gpu_enabled",
+      "default_value": false,
+      "validation": {
+        "type": "bool",
+        "required": false
+      }
+    }
+  ]
 }

src/tests/controllers/resources/test_resource.py

@@ -21,6 +21,7 @@ def translate_status(self, status):
         return status_list[status]
 
     def launch_job(self, job, workspace, resource_options):
+        self.validate_options(resource_options)
         return 0
 
     def get_resource_job(self, job):
@@ -32,12 +33,6 @@ def get_job_core_hours(self, job):
     def stop_job(self, job):
         return job.resource_job_id
 
-    def validate_options(self, resource_options):
-        # Should determine whether the requested options are valid for a resource
-        # Might be able to implement this at the abstract level once we've defined
-        #   a data model for resource options.
-        return True
-
     def translate_option_name(self, option):
         return None
 

src/tests/github_test_config.json

@@ -26,7 +26,12 @@
       "storage": "test_storage",
       "user_authentication": "test_auth",
       "passthrough_domain": "127.0.0.1:8000",
-      "connection_details": {}
+      "connection_details": {},
+      "parameter_mapping": {
+            "num_cpus": "cpus_per_task",
+            "memory_mb": "memory_per_node",
+            "time_limit_min": "time_limit"
+            }
     }
   },
   "available_job_types": {
@@ -38,5 +43,53 @@
         }
       }
     }
-  }
-}
+  },
+  "parameters": [
+    {
+      "display_name":  "Number of CPUs",
+      "description": "",
+      "variable_name": "num_cpus",
+      "default_value": 1,
+      "validation": {
+        "type": "int",
+        "min": 1,
+        "max": 4,
+        "required": false
+      }
+    },
+    {
+      "display_name": "Memory (MB)",
+      "description": "",
+      "variable_name": "memory_mb",
+      "default_value": 1024,
+      "validation": {
+        "type": "int",
+        "min": 1024,
+        "max": 8192,
+        "required": false
+      }
+    },
+    {
+      "display_name": "Time Limit (minutes)",
+      "description": "",
+      "variable_name": "time_limit_min",
+      "default_value": 180,
+      "validation": {
+        "type": "int",
+        "min": 1,
+        "max": 480,
+        "required": false
+      }
+    },
+    {
+      "display_name": "GPU Enabled",
+      "description": "",
+      "variable_name": "gpu_enabled",
+      "default_value": false,
+      "validation": {
+        "type": "bool",
+        "required": false
+      }
+    }
+  ]
+}

src/tests/unittests/test_views.py

@@ -583,6 +583,104 @@ def test_workspace_upload_multiple_files_put(self):
         test_file_1.close()
         test_file_2.close()
 
+    def test_workspace_custom_params_validation_good(self):
+        test_data_good = (
+            {"num_cpus": 1, "memory_mb": 1025},
+            {"time_limit_min": 480},
+            {"gpu_enabled": True},
+            {"unknown_param": "should_be_ignored"},
+            {},
+        )
+        self.client.force_authenticate(user=self.user)
+        body = {
+            "job_type": "test_job",
+            "job_details": {},
+        }
+        for test in test_data_good:
+            with self.subTest(test=test):
+                body["resource_options"] = test
+                response = self.client.put(
+                    reverse("workspaces_put_type", args=[self.workspace.id, "start"]),
+                    body,
+                )
+                self.assertValidResponse(
+                    response,
+                    status.HTTP_200_OK,
+                    success=True,
+                    message="Successful start.",
+                )
+
+    def test_workspace_custom_params_validation_bad(self):
+        test_data_bad = [
+            (
+                {"num_cpus": "one"},
+                {
+                    "msg": "[\"num_cpus: Value 'one' of type str does not match required type int. Skipping further validation of parameter num_cpus.\"]"
+                },
+            ),
+            (
+                {"memory_mb": 1},
+                {"msg": "[\"memory_mb: Value '1' not above minimum of 1024.\"]"},
+            ),
+            (
+                {"time_limit_min": 481},
+                {
+                    "msg": "[\"time_limit_min: Value '481' above maximum of 480.\"]",
+                },
+            ),
+        ]
+        self.client.force_authenticate(user=self.user)
+        body = {
+            "job_type": "test_job",
+            "job_details": {},
+        }
+        for test in test_data_bad:
+            with self.subTest(test=test):
+                body["resource_options"] = test[0]
+                response = self.client.put(
+                    reverse("workspaces_put_type", args=[self.workspace.id, "start"]),
+                    body,
+                )
+                self.assertValidResponse(
+                    response,
+                    status.HTTP_422_UNPROCESSABLE_ENTITY,
+                    success=False,
+                    message=f"Invalid resource options found: {test[1]['msg']}",
+                )
+
+    def test_workspace_custom_params_validation_required(self):
+        test_param = {
+            "display_name": "Test Param",
+            "description": "",
+            "variable_name": "test_param",
+            "validation": {"type": "str", "required": True},
+        }
+        apps.get_app_config("user_workspaces_server").parameters.append(test_param)
+        self.client.force_authenticate(user=self.user)
+        body = {
+            "job_type": "test_job",
+            "job_details": {},
+        }
+        response_bad = self.client.put(
+            reverse("workspaces_put_type", args=[self.workspace.id, "start"]),
+            body,
+        )
+        self.assertValidResponse(
+            response_bad,
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+            success=False,
+            message="Invalid resource options found: ['Missing required: test_param']",
+        )
+        body["resource_options"] = {"test_param": "test"}
+        response_good = self.client.put(
+            reverse("workspaces_put_type", args=[self.workspace.id, "start"]),
+            body,
+        )
+        self.assertValidResponse(response_good, status.HTTP_200_OK, success=True)
+        self.assertEqual(
+            apps.get_app_config("user_workspaces_server").parameters.pop(), test_param
+        )
+
 
 class WorkspaceDELETEAPITests(WorkspaceAPITestCase):
     def test_workspace_not_found_delete(self):
@@ -763,3 +861,16 @@ def test_job_types_get(self):
         self.client.force_authenticate(user=self.user)
         response = self.client.get(self.job_types_url)
         self.assertValidResponse(response, status.HTTP_200_OK, success=True, message="Successful.")
+
+
+class ParameterGETAPITest(UserWorkspacesAPITestCase):
+    parameters_url = reverse("parameters")
+    params_config = apps.get_app_config("user_workspaces_server").parameters
+
+    def test_parameters_get(self):
+        response = self.client.get(self.parameters_url)
+        self.assertValidResponse(response, status.HTTP_200_OK, success=True, message="Successful.")
+        self.assertContains(response, "data")
+        response_data = response.json()["data"]
+        self.assertIn("parameters", response_data)
+        self.assertEqual(response_data["parameters"], self.params_config)

src/user_workspaces_server/apps.py

@@ -22,6 +22,7 @@ def ready(self):
         config_storage = settings.UWS_CONFIG["available_storage"]
         config_resource = settings.UWS_CONFIG["available_resources"]
         config_job_types = settings.UWS_CONFIG["available_job_types"]
+        self.parameters = settings.UWS_CONFIG["parameters"]
 
         for (
             user_authentication_name,

src/user_workspaces_server/controllers/jobtypes/abstract_job.py

@@ -8,7 +8,7 @@ def __init__(self, config, job_details):
         self.script_template_name = None
 
     @abstractmethod
-    def get_script(self, template_params=None):
+    def get_script(self, template_params=None) -> str:
         pass
 
     @abstractmethod

src/user_workspaces_server/controllers/resources/abstract_resource.py

@@ -1,5 +1,13 @@
+import logging
 from abc import ABC, abstractmethod
 
+from user_workspaces_server.controllers.jobtypes.abstract_job import AbstractJob
+from user_workspaces_server.exceptions import ValidationException
+from user_workspaces_server.models import Job, Workspace
+from user_workspaces_server.validation.validate_job_params import ParamValidator
+
+logger = logging.getLogger(__name__)
+
 
 class AbstractResource(ABC):
     def __init__(self, config, resource_storage, resource_user_authentication):
@@ -9,38 +17,45 @@ def __init__(self, config, resource_storage, resource_user_authentication):
         self.passthrough_domain = config.get("passthrough_domain", "")
 
     @abstractmethod
-    def translate_status(self, status):
+    def translate_status(self, status: str) -> Job.Status:
         # Should take a string and translate it into local terminology.
         pass
 
     @abstractmethod
-    def launch_job(self, job, workspace, resource_options):
+    def launch_job(self, job: AbstractJob, workspace: Workspace, resource_options: dict) -> int:
         # Should return resource_job_id
         pass
 
     @abstractmethod
-    def get_resource_job(self, job):
+    def get_resource_job(self, job) -> dict:
         # Should get the resource's job information
         pass
 
     @abstractmethod
-    def get_job_core_hours(self, job):
+    def get_job_core_hours(self, job: Job) -> int:
         # Should return time in hours
         pass
 
     @abstractmethod
-    def stop_job(self, job):
+    def stop_job(self, job: Job) -> bool:
         # Should stop the job on the resource
         pass
 
-    @abstractmethod
-    def validate_options(self, resource_options):
-        # Should determine whether the requested options are valid for a resource
-        # Might be able to implement this at the abstract level once we've defined
-        #   a data model for resource options.
-        pass
+    def validate_options(self, resource_options: dict) -> bool:
+        validator = ParamValidator()
+        validator.validate(resource_options)
+        if validator.errors:
+            logging.error(f"Validation errors: {validator.errors}")
+            raise ValidationException(f"Invalid resource options found: {validator.errors}")
+        return validator.is_valid
 
-    @abstractmethod
-    def translate_options(self, resource_options):
-        # Should translate the options into a format that can be used by the resource
-        pass
+    def translate_option_name(self, option: str) -> str:
+        return self.config.get("parameter_mapping", {}).get(option)
+
+    def translate_options(self, resource_options: dict) -> dict:
+        translated_options = {}
+        for option_name, option_value in resource_options.items():
+            if updated_option_name := self.translate_option_name(option_name):
+                translated_options[updated_option_name] = option_value
+
+        return translated_options

src/user_workspaces_server/controllers/resources/local_resource.py

@@ -27,7 +27,7 @@ def translate_status(self, status):
 
         return status_list[status]
 
-    def launch_job(self, job, workspace):
+    def launch_job(self, job, workspace, resource_options):
         workspace_full_path = os.path.join(self.resource_storage.root_dir, workspace.file_path)
         job_full_path = os.path.join(workspace_full_path, f'.{job.job_details["id"]}')
         script_name = f"{str(time.time())}.sh"
@@ -89,13 +89,3 @@ def stop_job(self, job):
         except Exception as e:
             logger.error(repr(e))
             return False
-
-    def validate_options(self, resource_options):
-        # Should determine whether the requested options are valid for a resource
-        # Might be able to implement this at the abstract level once we've defined
-        #   a data model for resource options.
-        return True
-
-    def translate_options(self, resource_options):
-        # Should translate the options into a format that can be used by the resource
-        return {}