Patch: update support for fixed queue workflows (#169)

* first implementation

* docs

* pytest

* remove curated workflows CI as it is already deprecated

* update dockerfile

* accounting for archived pipelines

* improve message

* message

* message

* review suggestions

* removing extra space

.github/workflows/ci.yml

@@ -112,7 +112,6 @@ jobs:
         CLOUDOS_URL: "https://stg.sdlc.lifebit.ai"
       run: |
         cloudos workflow list --cloudos-url $CLOUDOS_URL --apikey $CLOUDOS_TOKEN --workspace-id $CLOUDOS_WORKSPACE_ID
-        cloudos workflow list --cloudos-url $CLOUDOS_URL --apikey $CLOUDOS_TOKEN --workspace-id $CLOUDOS_WORKSPACE_ID --curated
   project_list:
     runs-on: ubuntu-latest
     strategy:

CHANGELOG.md

@@ -1,5 +1,11 @@
 ## lifebit-ai/cloudos-cli: changelog
 
+## v2.11.1 (2024-10-22)
+
+### Fix
+
+- Updates queue support to disallow queue selection on fixed-queue workflows.
+
 ## v2.11.0 (2024-04-16)
 
 - Now, the default `cloudos job run` command will save job process logs. To prevent saving process logs, you can use the new flag `--do-not-save-logs`.

Dockerfile

@@ -1,8 +1,8 @@
 # Full contents of Dockerfile
 
-FROM continuumio/miniconda3:23.10.0-1
+FROM continuumio/miniconda3:24.7.1-0
 LABEL name="quay.io/lifebitaiorg/cloudos-cli" \
-      description="The cloudos-py docker container" \
+      description="The cloudos-cli docker container" \
       maintainer="David Pineyro <[email protected]>"
 
 # Use the base conda env to not be reliant on conda activate when using pip

README.md

@@ -11,6 +11,7 @@ The package requires Python >= 3.7 and the following python packages:
 ```
 click>=8.0.1
 pandas>=1.3.4
+numpy==1.26.4
 requests>=2.26.0
 ```
 
@@ -722,9 +723,15 @@ Executing list...
 	Job queue list saved to available_queues.json
 ```
 
-> NOTE: the job name that is visible in CloudOS and has to be used in combination with `--job-queue` parameter is
+> NOTE: the queue name that is visible in CloudOS and has to be used in combination with `--job-queue` parameter is
 the one in `label` field.
 
+##### Job queues for platform workflows
+
+Platform workflows, i.e., those provided by CloudOS in your workspace as modules, run on separated and specific AWS batch queues.
+Therefore, CloudOS will automatically assign the valid queue and the user should not specify any queue using the `--job-queue` paramater.
+Any attempt of using this parameter will be ignored. Examples of such platform workflows are "System Tools" and "Data Factory" workflows.
+
 ### WDL pipeline support
 
 #### Cromwell server managing

cloudos/__init__.py

@@ -8,4 +8,4 @@
 from .clos import Cloudos
 from ._version import __version__
 
-__all__ = ['jobs', 'utils', 'clos', 'cohorts']
+__all__ = ['jobs', 'utils', 'clos', 'queue']

cloudos/__main__.py

@@ -299,6 +299,7 @@ def run(apikey,
         print('\t...Detecting workflow type')
     cl = Cloudos(cloudos_url, apikey, cromwell_token)
     workflow_type = cl.detect_workflow(workflow_name, workspace_id, verify_ssl)
+    is_module = cl.is_module(workflow_name, workspace_id, verify_ssl)
     if execution_platform == 'hpc' and workflow_type == 'wdl':
         raise ValueError(f'The workflow {workflow_name} is a WDL workflow. ' +
                          'WDL is not supported on HPC execution platform.')
@@ -343,10 +344,17 @@ def run(apikey,
         print('\tThe following Job object was created:')
         print('\t' + str(j))
         print('\t...Sending job to CloudOS\n')
-    queue = Queue(cloudos_url=cloudos_url, apikey=apikey, cromwell_token=cromwell_token,
-                  workspace_id=workspace_id, verify=verify_ssl)
-    job_queue_id = queue.fetch_job_queue_id(workflow_type=workflow_type, batch=batch,
-                                            job_queue=job_queue)
+    if is_module:
+        if job_queue is not None:
+            print(f'\tIgnoring job queue "{job_queue}" for ' +
+                  f'Platform Workflow "{workflow_name}". Platform Workflows ' +
+                  'use their own predetermined queues.')
+        job_queue_id = None
+    else:
+        queue = Queue(cloudos_url=cloudos_url, apikey=apikey, cromwell_token=cromwell_token,
+                      workspace_id=workspace_id, verify=verify_ssl)
+        job_queue_id = queue.fetch_job_queue_id(workflow_type=workflow_type, batch=batch,
+                                                job_queue=job_queue)
     j_id = j.send_job(job_config=job_config,
                       parameter=parameter,
                       git_commit=git_commit,

cloudos/_version.py

@@ -1 +1 @@
-__version__ = '2.11.0'
+__version__ = '2.11.1'

cloudos/clos.py

@@ -418,6 +418,7 @@ def process_workflow_list(r, all_fields=False):
         """
         COLUMNS = ['_id',
                    'name',
+                   'isModule',
                    'archived.status',
                    'mainFile',
                    'workflowType',
@@ -459,14 +460,49 @@ def detect_workflow(self, workflow_name, workspace_id, verify=True):
         """
         my_workflows_r = self.get_workflow_list(workspace_id, verify=verify)
         my_workflows = self.process_workflow_list(my_workflows_r)
-        wt_all = my_workflows.loc[my_workflows['name'] == workflow_name, 'workflowType']
+        wt_all = my_workflows.loc[
+            (my_workflows['name'] == workflow_name) & (my_workflows['archived.status'] == False),
+            'workflowType']
         if len(wt_all) == 0:
             raise ValueError(f'No workflow found with name: {workflow_name}')
         wt = wt_all.unique()
         if len(wt) > 1:
             raise ValueError(f'More than one workflow type detected for {workflow_name}: {wt}')
         return str(wt[0])
 
+    def is_module(self, workflow_name, workspace_id, verify=True):
+        """Detects whether the workflow is a system module or not.
+
+        System modules use fixed queues, so this check is important to
+        properly manage queue selection.
+
+        Parameters
+        ----------
+        workflow_name : string
+            Name of the workflow.
+        workspace_id : string
+            The CloudOS workspace id from to collect the workflows.
+        verify: [bool|string]
+            Whether to use SSL verification or not. Alternatively, if
+            a string is passed, it will be interpreted as the path to
+            the SSL certificate file.
+
+        Returns
+        -------
+        bool
+            True, if the workflow is a system module, false otherwise.
+        """
+        my_workflows_r = self.get_workflow_list(workspace_id, verify=verify)
+        my_workflows = self.process_workflow_list(my_workflows_r)
+        is_module = my_workflows.loc[
+            (my_workflows['name'] == workflow_name) & (my_workflows['archived.status'] == False),
+            'isModule']
+        if len(is_module) == 0:
+            raise ValueError(f'No workflow found with name: {workflow_name}')
+        if len(is_module) > 1:
+            raise ValueError(f'More than one workflow found with name: {workflow_name}')
+        return is_module.values[0]
+
     def get_project_list(self, workspace_id, verify=True):
         """Get all the project from a CloudOS workspace.
 

environment.yml

@@ -8,3 +8,4 @@ dependencies:
     - requests
     - click
     - pandas
+    - numpy==1.26.4

requirements.txt

@@ -2,6 +2,7 @@ click>=8.0.1
 requests>=2.26.0
 requests-mock>=1.9.3
 pandas>=1.3.4
+numpy==1.26.4
 pytest>=6.2.5
 responses>=0.21.0
-mock>=3.0.5
+mock>=3.0.5

tests/test_clos/test_is_module.py

@@ -0,0 +1,41 @@
+"""Pytest for method Cloudos.is_module"""
+import mock
+import responses
+from responses import matchers
+from cloudos.clos import Cloudos
+from tests.functions_for_pytest import load_json_file
+
+INPUT = "tests/test_data/process_workflow_list_initial_request.json"
+APIKEY = 'vnoiweur89u2ongs'
+CLOUDOS_URL = 'http://cloudos.lifebit.ai'
+WORKSPACE_ID = 'lv89ufc838sdig'
+
+
+@mock.patch('cloudos.clos', mock.MagicMock())
+@responses.activate
+def test_is_module():
+    """
+    Test 'is_module' to work as intended
+    API request is mocked and replicated with json files
+    """
+    json_data = load_json_file(INPUT)
+    params = {"teamId": WORKSPACE_ID, "apikey": APIKEY}
+    header = {
+        "Accept": "application/json, text/plain, */*",
+        "Content-Type": "application/json;charset=UTF-8"
+    }
+    search_str = f"teamId={WORKSPACE_ID}&apikey={APIKEY}"
+    # mock GET method with the .json
+    responses.add(
+            responses.GET,
+            url=f"{CLOUDOS_URL}/api/v1/workflows?{search_str}",
+            body=json_data,
+            headers=header,
+            match=[matchers.query_param_matcher(params)],
+            status=200)
+    # start cloudOS service
+    clos = Cloudos(apikey=APIKEY, cromwell_token=None, cloudos_url=CLOUDOS_URL)
+    # get mock response
+    response = clos.is_module(workspace_id=WORKSPACE_ID,
+                              workflow_name="multiqc")
+    assert response

tests/test_data/process_workflow_list_initial_request.json

@@ -292,7 +292,7 @@
         "isPredefined": false,
         "isCurated": false,
         "isFeatured": false,
-        "isModule": false,
+        "isModule": true,
         "isPublic": false,
         "priceAmount": 0,
         "priceUnit": "PER_SAMPLE",
@@ -309,4 +309,4 @@
         "workflowType": "docker",
         "containerName": "ewels/multiqc"
     }
-]
+]

tests/test_data/process_workflow_list_results.csv

@@ -1,4 +1,4 @@
-_id,name,archived.status,mainFile,workflowType,repository.name,repository.platform,repository.url,repository.isPrivate
-XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX
-XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX
-XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX
+_id,name,isModule,archived.status,mainFile,workflowType,repository.name,repository.platform,repository.url,repository.isPrivate
+XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX
+XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX
+XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX