First untested attempt at CLI plugin for LSF

2026-10-24 12:13:33 +08:00 · 2018-09-27 18:44:21 +01:00
parent 872b6f57ed
commit ebcad566da
1 changed files with 102 additions and 0 deletions
@@ -0,0 +1,102 @@
+# A simple CLI runner for slurm that can be used when running Galaxy from a
+# non-submit host and using a Slurm cluster.
+from logging import getLogger
+
+try:
+    from galaxy.model import Job
+    job_states = Job.states
+except ImportError:
+    # Not in Galaxy, map Galaxy job states to Pulsar ones.
+    from pulsar.util import enum
+    job_states = enum(RUNNING='running', OK='complete', QUEUED='queued', ERROR="failed")
+
+from ..job import BaseJobExec
+
+log = getLogger(__name__)
+
+argmap = {
+    'memory': '-M',
+    'cores': '-n',
+    'queue': '-q',
+    'working_dir': '-cwd'
+}
+
+
+class LSF(BaseJobExec):
+
+    def __init__(self, **params):
+        self.params = {}
+        for k, v in params.items():
+            self.params[k] = v
+
+    def job_script_kwargs(self, ofile, efile, job_name):
+        scriptargs = {'-o': ofile,
+                      '-e': efile,
+                      '-J': job_name}
+
+        # Map arguments using argmap.
+        for k, v in self.params.items():
+            if k == 'plugin':
+                continue
+            try:
+                if not k.startswith('-'):
+                    k = argmap[k]
+                if k == 'memory':
+                    # Memory requires both -m and -R rusage[mem=v] request
+                    scriptargs['-R'] = "\"rusage[mem=%s]\"" % v
+                scriptargs[k] = v
+            except Exception:
+                log.warning('Unrecognized long argument passed to LSF CLI plugin: %s' % k)
+
+        # Generated template.
+        template_scriptargs = ''
+        for k, v in scriptargs.items():
+            template_scriptargs += '#BSUB %s %s\n' % (k, v)
+        return dict(headers=template_scriptargs)
+
+    def submit(self, script_file):
+        return 'bsub <%s' % script_file
+
+    def delete(self, job_id):
+        return 'bstop %s' % job_id
+
+    def get_status(self, job_ids=None):
+        return "bjobs -o \"id stat delimiter='^'\" -noheader"  # check this
+
+
+    def get_single_status(self, job_id):
+        return "bjobs -o stat -noheader " + job_id
+
+    def parse_status(self, status, job_ids):
+        # Get status for each job, skipping header.
+        rval = {}
+        for line in status.splitlines()[1:]:
+            id, state = line.split('^')
+            if id in job_ids:
+                # map job states to Galaxy job states.
+                rval[id] = self._get_job_state(state)
+        return rval
+
+    def parse_single_status(self, status, job_id):
+        if "is not found" in status:
+            raise Exception("Job %s not found in LSF" % job_id)
+        return self._get_job_state(status)
+
+    def _get_job_state(self, state):
+        # based on:
+        # https://www.ibm.com/support/knowledgecenter/en/SSETD4_9.1.3/lsf_admin/job_state_lsf.html
+        try:
+            return {
+                'EXIT': job_states.ERROR,
+                'RUN': job_states.RUNNING,
+                'PEND': job_states.QUEUED,
+                'DONE': job_states.OK,
+                'PSUSP': job_states.ERROR,
+                'USUSP': job_states.ERROR,
+                'SSUSP': job_states.ERROR
+            }.get(state)
+        except KeyError:
+            raise KeyError("Failed to map LSF status code [%s] to job state." % state)
+
+
+__all__ = ('LSF',)