Feat: adds private docker container support

CHANGELOG.md

@@ -1,5 +1,11 @@
 ## lifebit-ai/cloudos-cli: changelog
 
+## v2.16.0 (2025-01-21)
+
+### Feature
+
+- Adds the new parameter `--use_private_docker_repository` to launch jobs using private docker images from a linked docker.io account.
+
 ## v2.15.0 (2025-01-16)
 
 ### Feature

README.md

@@ -162,6 +162,10 @@ Options:
                                   (For no cost limit please use -1).
   --accelerate-file-staging       Enables AWS S3 mountpoint for quicker file
                                   staging.
+  --use-private-docker-repository
+                                  Allows to use private docker repository for
+                                  running jobs. The Docker user account has to
+                                  be already linked to CloudOS.
   --verbose                       Whether to print information messages or
                                   not.
   --request-interval INTEGER      Time interval to request (in seconds) the

cloudos/__main__.py

@@ -214,6 +214,10 @@ def queue():
 @click.option('--accelerate-file-staging',
               help='Enables AWS S3 mountpoint for quicker file staging.',
               is_flag=True)
+@click.option('--use-private-docker-repository',
+              help=('Allows to use private docker repository for running jobs. The Docker user ' +
+                    'account has to be already linked to CloudOS.'),
+              is_flag=True)
 @click.option('--verbose',
               help='Whether to print information messages or not.',
               is_flag=True)
@@ -261,6 +265,7 @@ def run(apikey,
         hpc_id,
         cost_limit,
         accelerate_file_staging,
+        use_private_docker_repository,
         verbose,
         request_interval,
         disable_ssl_verification,
@@ -386,6 +391,24 @@ def run(apikey,
                       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 use_private_docker_repository:
+        if is_module:
+            print(f'[Message] Workflow "{workflow_name}" is a CloudOS module. ' +
+                  'Option --use-private-docker-repository will be ignored.')
+            docker_login = False
+        else:
+            me = j.get_user_info(verify=verify_ssl)['dockerRegistriesCredentials']
+            if len(me) == 0:
+                raise Exception('User private Docker repository has been selected but your user ' +
+                                'credentials have not been configured yet. Please, link your ' +
+                                'Docker account to CloudOS before using ' +
+                                '--use-private-docker-repository option.')
+            print('[Message] Use private Docker repository has been selected. A custom job ' +
+                  'queue to support private Docker containers and/or Lustre FSx will be created for ' +
+                  'your job. The selected job queue will serve as a template.')
+            docker_login = True
+    else:
+        docker_login = False
     if nextflow_version == 'latest':
         nextflow_version = '24.04.4'
         print('[Message] You have specified Nextflow version \'latest\'. The workflow will use the ' +
@@ -417,6 +440,7 @@ def run(apikey,
                       cromwell_id=cromwell_id,
                       cost_limit=cost_limit,
                       use_mountpoints=use_mountpoints,
+                      docker_login=docker_login,
                       verify=verify_ssl)
     print(f'\tYour assigned job id is: {j_id}\n')
     j_url = f'{cloudos_url}/app/jobs/{j_id}'

cloudos/_version.py

@@ -1 +1 @@
-__version__ = '2.15.0'
+__version__ = '2.16.0'

cloudos/clos.py

@@ -699,3 +699,28 @@ def workflow_import(self, workspace_id, workflow_url, workflow_name,
             raise BadRequestException(r)
         content = json.loads(r.content)
         return content['_id']
+
+    def get_user_info(self, verify=True):
+        """Gets user information from users/me endpoint
+
+        Parameters
+        ----------
+        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
+        -------
+        r : requests.models.Response.content
+            The server response content
+        """
+        headers = {
+            "Content-type": "application/json",
+            "apikey": self.apikey
+        }
+        r = retry_requests_get("{}/api/v1/users/me".format(self.cloudos_url),
+                               headers=headers, verify=verify)
+        if r.status_code >= 400:
+            raise BadRequestException(r)
+        return json.loads(r.content)

cloudos/jobs/job.py

@@ -199,7 +199,8 @@ def convert_nextflow_to_json(self,
                                  workflow_type,
                                  cromwell_id,
                                  cost_limit,
-                                 use_mountpoints):
+                                 use_mountpoints,
+                                 docker_login):
         """Converts a nextflow.config file into a json formatted dict.
 
         Parameters
@@ -260,6 +261,8 @@ def convert_nextflow_to_json(self,
             Job cost limit. -1 means no cost limit.
         use_mountpoints : bool
             Whether to use or not AWS S3 mountpoint for quicker file staging.
+        docker_login : bool
+            Whether to use private docker images, provided the users have linked their docker.io accounts.
 
         Returns
         -------
@@ -385,7 +388,7 @@ def convert_nextflow_to_json(self,
             "resumable": resumable,
             "saveProcessLogs": save_logs,
             "batch": {
-                "dockerLogin": False,
+                "dockerLogin": docker_login,
                 "enabled": batch,
                 "jobQueue": job_queue_id
             },
@@ -436,6 +439,7 @@ def send_job(self,
                  cromwell_id=None,
                  cost_limit=30.0,
                  use_mountpoints=False,
+                 docker_login=False,
                  verify=True):
         """Send a job to CloudOS.
 
@@ -493,6 +497,8 @@ def send_job(self,
             Job cost limit. -1 means no cost limit.
         use_mountpoints : bool
             Whether to use or not AWS S3 mountpoint for quicker file staging.
+        docker_login : bool
+            Whether to use private docker images, provided the users have linked their docker.io accounts.
         verify: [bool|string]
             Whether to use SSL verification or not. Alternatively, if
             a string is passed, it will be interpreted as the path to
@@ -536,7 +542,8 @@ def send_job(self,
                                                workflow_type,
                                                cromwell_id,
                                                cost_limit,
-                                               use_mountpoints)
+                                               use_mountpoints,
+                                               docker_login)
         r = retry_requests_post("{}/api/v1/jobs?teamId={}".format(cloudos_url,
                                                             workspace_id),
                                 data=json.dumps(params), headers=headers, verify=verify)

cloudos/queue/queue.py

@@ -128,7 +128,7 @@ def fetch_job_queue_id(self, workflow_type, batch=True, job_queue=None):
             default_queue_name = available_queues[-1]['label']
             queue_as_default = 'most recent suitable'
         if job_queue is None:
-            print(f'[Message] No job_queue was specified, using the {queue_as_default} queue: ' +
+            print(f'[Message] No job queue was specified, using the {queue_as_default} queue: ' +
                   f'{default_queue_name}.')
             return default_queue_id
         selected_queue = [q for q in available_queues if q['label'] == job_queue]

tests/test_clos/test_get_user_info.py

@@ -0,0 +1,68 @@
+import mock
+import json
+import pytest
+import responses
+from cloudos.clos import Cloudos
+from cloudos.utils.errors import BadRequestException
+
+APIKEY = 'vnoiweur89u2ongs'
+CLOUDOS_URL = 'http://cloudos.lifebit.ai'
+
+
+@mock.patch('cloudos.clos', mock.MagicMock())
+@responses.activate
+def test_get_user_info_correct_response():
+    """
+    Test 'get_user_info' to work as intended
+    """
+    body = json.dumps({"dockerRegistriesCredentials": []})
+    header = {
+        "Accept": "application/json, text/plain, */*",
+        "Content-Type": "application/json;charset=UTF-8",
+        "apikey": APIKEY
+    }
+    # mock GET method with the .json
+    responses.add(
+            responses.GET,
+            body=body,
+            url=f"{CLOUDOS_URL}/api/v1/users/me",
+            headers=header,
+            status=200)
+    # start cloudOS service
+    clos = Cloudos(apikey=APIKEY, cromwell_token=None,
+                   cloudos_url=CLOUDOS_URL)
+    # get mock response
+    response = clos.get_user_info()
+    # check the response
+    assert response['dockerRegistriesCredentials'] == []
+
+
+@mock.patch('cloudos.clos', mock.MagicMock())
+@responses.activate
+def test_get_user_info_incorrect_response():
+    """
+    Test 'get_user_info' to fail with '400' response
+    """
+    # prepare error message
+    error_message = {"statusCode": 400, "code": "BadRequest",
+                     "message": "Bad Request.", "time": "2022-11-23_17:31:07"}
+    error_json = json.dumps(error_message)
+    header = {
+        "Accept": "application/json, text/plain, */*",
+        "Content-Type": "application/json;charset=UTF-8",
+        "apikey": APIKEY
+    }
+    # mock GET method with the .json
+    responses.add(
+            responses.GET,
+            url=f"{CLOUDOS_URL}/api/v1/users/me",
+            body=error_json,
+            headers=header,
+            status=400)
+    # raise 400 error
+    with pytest.raises(BadRequestException) as error:
+        # check if it failed
+        clos = Cloudos(apikey=APIKEY, cromwell_token=None,
+                       cloudos_url=CLOUDOS_URL)
+        clos.get_user_info()
+    assert "Server returned status 400." in (str(error))

tests/test_jobs/test_convert_nextflow_to_json.py

@@ -29,7 +29,8 @@
     "workflow_type": 'nextflow',
     "cromwell_id": None,
     "cost_limit": -1,
-    "use_mountpoints": False
+    "use_mountpoints": False,
+    "docker_login": False
 }
 
 
@@ -58,7 +59,8 @@ def test_convert_nextflow_to_json_output_correct():
         workflow_type=param_dict["workflow_type"],
         cromwell_id=param_dict["cromwell_id"],
         cost_limit=param_dict["cost_limit"],
-        use_mountpoints=param_dict["use_mountpoints"]
+        use_mountpoints=param_dict["use_mountpoints"],
+        docker_login=param_dict["docker_login"]
         )
     with open(actual_json_file) as json_data:
         correct_json = json.load(json_data)
@@ -92,7 +94,8 @@ def test_convert_nextflow_to_json_badly_formed_config():
             workflow_type=param_dict["workflow_type"],
             cromwell_id=param_dict["cromwell_id"],
             cost_limit=param_dict["cost_limit"],
-            use_mountpoints=param_dict["use_mountpoints"]
+            use_mountpoints=param_dict["use_mountpoints"],
+            docker_login=param_dict["docker_login"]
             )
         print(str(excinfo.value))
     assert "Please, specify your parameters in\