Add singularity and slurm support

.gitignore

@@ -4,10 +4,13 @@
 /docs/_build/
 /tmp/
 __pycache__/
-/girder_worker.egg-info/
+*.egg-info/
 /girder-worker-*.tar.gz
 *.retry
 .vagrant
 .tox/
 .cache/
 htmlcov/
+build/
+.eggs
+

examples/plugin_example/gwexample/analyses/tasks.py

@@ -1,4 +1,3 @@
-from girder_worker.app import app
 from girder_worker_utils import types
 from girder_worker_utils.decorators import argument
 

girder_worker/app.py

@@ -132,7 +132,8 @@ def gw_task_prerun(task=None, sender=None, task_id=None,
             raise
 
     try:
-        task.girder_client = GirderClient(apiUrl=task.request.girder_api_url)
+        # task.girder_client = GirderClient(apiUrl=task.request.girder_api_url)
+        task.girder_client = GirderClient(apiUrl='http://0.0.0.0:8101/api/v1')
         task.girder_client.token = task.request.girder_client_token
     except AttributeError:
         task.girder_client = None

girder_worker/docker/io/__init__.py

@@ -191,6 +191,7 @@ class FileDescriptorReader(StreamReader):
     """
     Reader to read from a file descriptor.
     """
+
     def __init__(self, fd):
         self._fd = fd
 
@@ -208,6 +209,7 @@ class FileDescriptorWriter(StreamWriter):
     """
     Writer to write to a file descriptor.
     """
+
     def __init__(self, fd):
         self._fd = fd
 
@@ -225,6 +227,7 @@ class StdStreamWriter(StreamWriter):
     """
     Writer for write to stdout and stderr.
     """
+
     def __init__(self, stream):
         self._stream = stream
 
@@ -240,6 +243,7 @@ class NamedPipe:
     """
     A named pipe.
     """
+
     def __init__(self, path):
         self.path = path
         self._fd = None
@@ -267,6 +271,7 @@ class NamedPipeReader(FileDescriptorReader):
     """
     Reader to read from a named pipe.
     """
+
     def __init__(self, pipe, container_path=None):
         super().__init__(None)
         self._pipe = pipe
@@ -290,6 +295,7 @@ class NamedPipeWriter(FileDescriptorWriter):
     """
     Write to write to a named pipe.
     """
+
     def __init__(self, pipe, container_path=None):
         super().__init__(None)
         self._pipe = pipe

girder_worker/docker/io/girder.py

@@ -5,6 +5,7 @@ class GirderFileStreamReader(StreamReader):
     """
     Stream a file from Girder.
     """
+
     def __init__(self, client, file_id):
         """
         :param client: The GirderClient instance to use.

girder_worker/docker/tasks/__init__.py

@@ -5,16 +5,19 @@
 import sys
 import threading
 import time
+
 try:
     import docker
-    from docker.errors import DockerException, APIError, InvalidVersion
-    from girder_worker.docker import nvidia
+    from docker.errors import APIError, DockerException, InvalidVersion
     from requests.exceptions import ReadTimeout
+
+    from girder_worker.docker import nvidia
 except ImportError:
     # These imports will not be available on the girder side.
     pass
-from girder_worker.app import app, Task
+
 from girder_worker import logger
+from girder_worker.app import Task, app
 from girder_worker.docker import utils
 from girder_worker.docker.stream_adapter import DockerStreamPushAdapter
 from girder_worker.docker.io import (
@@ -24,6 +27,7 @@
     FDStreamConnector,
     StdStreamWriter
 )
+
 from girder_worker.docker.transforms import (
     ContainerStdErr,
     ContainerStdOut,
@@ -145,6 +149,7 @@ class _SocketReader(FileDescriptorReader):
     with python 2 attach_socket(...) returns a socket like object, with python 3
     it returns an instance of SocketIO.
     """
+
     def __init__(self, socket):
         self._socket = socket
 
@@ -424,7 +429,6 @@ def _docker_run(task, image, pull_image=True, entrypoint=None, container_args=No
     results = []
     if hasattr(task.request, 'girder_result_hooks'):
         results = (None,) * len(task.request.girder_result_hooks)
-
     return results
 
 

girder_worker/docker/transforms/__init__.py

@@ -22,6 +22,7 @@ class HostStdOut(Transform):
     Represents the standard output stream on the host machine. Can be used with
     :py:class:`girder_worker.docker.transforms.Connect` to write text to stdout.
     """
+
     def transform(self, **kwargs):
         from girder_worker.docker.io import (
             StdStreamWriter
@@ -34,6 +35,7 @@ class HostStdErr(Transform):
     Represents the standard error stream on the host machine. Can be used with
     :py:class:`girder_worker.docker.transforms.Connect` to write text to stderr.
     """
+
     def transform(self, **kwargs):
         from girder_worker.docker.io import (
             StdStreamWriter
@@ -47,6 +49,7 @@ class ContainerStdOut(Transform):
     :py:class:`girder_worker.docker.transforms.Connect` to redirect the containers
     standard output to another stream.
     """
+
     def transform(self, **kwargs):
         return self
 
@@ -61,6 +64,7 @@ class ContainerStdErr(Transform):
     :py:class:`girder_worker.docker.transforms.Connect` to redirect the containers
     standard error to another stream.
     """
+
     def transform(self, **kwargs):
         return self
 
@@ -81,6 +85,7 @@ class BindMountVolume(Transform):
     :param mode: The mounting mode
     :type mode: str
     """
+
     def __init__(self, host_path, container_path, mode='rw'):
         self._host_path = host_path
         self._container_path = container_path
@@ -156,6 +161,7 @@ class TemporaryVolume(_TemporaryVolumeBase, metaclass=_TemporaryVolumeMetaClass)
     """
     # Note that this mode is explicitly set with os.chmod. What you
     # set, is what you get - no os.makedirs umask shenanigans.
+
     def __init__(self, host_dir=None, mode=0o777):
         super().__init__(None, None)
         self.host_dir = host_dir
@@ -179,6 +185,7 @@ class _DefaultTemporaryVolume(TemporaryVolume):
     containing information about the actual default temporary volume associated with the
     task. The place holder then delegates all functionality to this instance.
     """
+
     def transform(self, _default_temp_volume=None, **kwargs):
         self._instance = _default_temp_volume
         self._transformed = True
@@ -253,6 +260,7 @@ class NamedInputPipe(NamedPipeBase):
         the volume will be used when creating the pipe. The default location is
         :py:obj:`girder_worker.docker.transforms.TemporaryVolume.default`
     """
+
     def __init__(self, name,  container_path=None, host_path=None, volume=TemporaryVolume.default):
         super().__init__(name, container_path, host_path, volume)
 
@@ -283,6 +291,7 @@ class NamedOutputPipe(NamedPipeBase):
         the volume will be use when creating the pipe. The default location is
         :py:attr:`girder_worker.docker.transforms.TemporaryVolume.default`
     """
+
     def __init__(self, name, container_path=None, host_path=None, volume=TemporaryVolume.default):
         super().__init__(name, container_path, host_path, volume)
 
@@ -308,11 +317,13 @@ class VolumePath(Transform):
         :py:attr:`girder_worker.docker.transforms.TemporaryVolume.default`
     :type volume: :py:class:`girder_worker.docker.transforms.BindMountVolume`
     """
+
     def __init__(self, filename, volume=TemporaryVolume.default):
         if os.path.isabs(filename):
             raise Exception('VolumePath paths must be relative to a volume (%s).' % filename)
-
-        self.filename = filename
+        # Modify filename for cli_run
+        # self.filename = filename
+        self.filename = filename.replace(' ', '_')
         self._volume = volume
 
     def transform(self, *pargs, **kwargs):
@@ -344,6 +355,7 @@ class Connect(Transform):
         :py:class:`girder_worker.docker.transforms.HostStdOut` or
         :py:class:`girder_worker.docker.transforms.HostStdErr`
     """
+
     def __init__(self, input, output):
         super().__init__()
         self._input = input
@@ -380,6 +392,7 @@ class ChunkedTransferEncodingStream(Transform):
     :param headers: HTTP headers to send.
     :type header: dict
     """
+
     def __init__(self, url, headers={}, **kwargs):
         self.url = url
         self.headers = headers

girder_worker/docker/transforms/girder.py

@@ -45,6 +45,7 @@ class ProgressPipe(Transform):
     :param volume: The bind mount volume where the underlying named pipe will reside.
     :type volume: :py:class:`girder_worker.docker.transforms.BindMountVolume`
     """
+
     def __init__(self, name='.girder_progress', volume=TemporaryVolume.default):
         self.name = name
         self._volume = volume
@@ -70,6 +71,7 @@ class GirderFileIdToStream(GirderClientTransform):
     :param _id: The Girder file ID.
     :type _id: str or ObjectId
     """
+
     def __init__(self, _id, **kwargs):
         super().__init__(**kwargs)
         self.file_id = _id
@@ -93,6 +95,7 @@ class GirderFileIdToVolume(GirderClientTransform):
     :param filename: Alternate name for the file. Default is to use the name from Girder.
     :type filename: str
     """
+
     def __init__(self, _id, volume=TemporaryVolume.default, filename=None, **kwargs):
         super().__init__(**kwargs)
         self._file_id = str(_id)
@@ -153,6 +156,7 @@ class GirderFolderIdToVolume(GirderClientTransform):
     :param folder_name: Alternate name for the directory. Default is to use the name from Girder.
     :type folder_name: str
     """
+
     def __init__(self, _id, volume=TemporaryVolume.default, folder_name=None, **kwargs):
         super().__init__(**kwargs)
         self._folder_id = str(_id)
@@ -206,6 +210,7 @@ class GirderItemIdToVolume(GirderClientTransform):
     :param item_name: Alternate name for the file. Default is to use the name from Girder.
     :type item_name: str
     """
+
     def __init__(self, _id, volume=TemporaryVolume.default, **kwargs):
         super().__init__(**kwargs)
         self._item_id = str(_id)
@@ -257,6 +262,7 @@ class GirderUploadVolumePathToItem(GirderUploadToItem):
     :param delete_file: Whether to delete the file afterward.
     :type delete_file: bool
     """
+
     def __init__(self, volumepath, item_id, delete_file=False, **kwargs):
         item_id = str(item_id)
         super().__init__(item_id, delete_file, **kwargs)
@@ -281,6 +287,7 @@ class GirderUploadVolumePathToFolder(GirderUploadToFolder):
     :param delete_file: Whether to delete the data afterward.
     :type delete_file: bool
     """
+
     def __init__(self, volumepath, folder_id, delete_file=False, **kwargs):
         super().__init__(str(folder_id), delete_file, **kwargs)
         self._volumepath = volumepath
@@ -311,6 +318,7 @@ class GirderUploadVolumePathJobArtifact(GirderUploadJobArtifact):
         fails. This can be used to debug failed ``docker_run`` tasks.
     :type upload_on_exception: bool
     """
+
     def __init__(self, volumepath, job_id=None, name=None, upload_on_exception=False, **kwargs):
         if job_id is not None:
             job_id = str(job_id)

girder_worker/entrypoint.py

@@ -1,7 +1,7 @@
 from importlib import import_module
+
 import celery
 from girder_worker_utils import decorators
-
 from stevedore import extension
 
 

girder_worker/girder_plugin/web_client/templates/configView.pug

@@ -19,7 +19,7 @@ form#g-worker-settings-form(role="form")
   .checkbox
     label.control-label(for="g-worker-direct-path")
       input#g-worker-direct-path(type="checkbox")
-      span When possible, send local file paths to the worker to avoid downloading files
+      span When possible, send local file paths to the worker to avoid downloading file
 
   p#g-worker-settings-error-message.g-validation-failed-message
   input.btn.btn-sm.btn-primary(type="submit", value="Save")

girder_worker/girder_plugin/web_client/views/ConfigView.js

@@ -41,7 +41,7 @@ var ConfigView = View.extend({
                     'worker.api_url',
                     'worker.broker',
                     'worker.backend',
-                    'worker.direct_path'
+                    'worker.direct_path',
                 ])
             }
         }).done((resp) => {

girder_worker/singularity/girder_worker_singularity/__init__.py

@@ -0,0 +1,10 @@
+from girder_worker import GirderWorkerPluginABC
+
+
+class SingularityPlugin(GirderWorkerPluginABC):
+
+    def __init__(self, app, *args, **kwargs):
+        self.app = app
+
+    def task_imports(self):
+        return ['girder_worker_singularity.tasks']

girder_worker/singularity/girder_worker_singularity/girder_plugin.py

@@ -0,0 +1,8 @@
+from girder.plugin import GirderPlugin
+
+
+class WorkerSingularityPlugin(GirderPlugin):
+    DISPLAY_NAME = 'Worker Singularity'
+
+    def load(self, info):
+        pass