From febadeae65cb61f52a497f2ddc2e3c9ec64e6ce3 Mon Sep 17 00:00:00 2001 From: Nate Coraor Date: Wed, 6 Apr 2016 15:32:37 -0400 Subject: [PATCH 1/3] Convert the DRMAA runner to use Pulsar's DRMAA session wrapper --- lib/galaxy/jobs/runners/drmaa.py | 46 +++++++++++--------------------- 1 file changed, 16 insertions(+), 30 deletions(-) diff --git a/lib/galaxy/jobs/runners/drmaa.py b/lib/galaxy/jobs/runners/drmaa.py index c6928b398d3..05620d0a544 100644 --- a/lib/galaxy/jobs/runners/drmaa.py +++ b/lib/galaxy/jobs/runners/drmaa.py @@ -15,6 +15,7 @@ from galaxy.jobs import JobDestination from galaxy.jobs.handler import DEFAULT_JOB_PUT_FAILURE_MESSAGE from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner from galaxy.util import asbool +from pulsar.managers.util.drmaa import DrmaaSessionFactory drmaa = None @@ -22,9 +23,6 @@ log = logging.getLogger( __name__ ) __all__ = [ 'DRMAAJobRunner' ] -DRMAA_jobTemplate_attributes = [ 'args', 'remoteCommand', 'outputPath', 'errorPath', 'nativeSpecification', - 'workingDirectory', 'jobName', 'email', 'project' ] - class DRMAAJobRunner( AsynchronousJobRunner ): """ @@ -81,8 +79,8 @@ class DRMAAJobRunner( AsynchronousJobRunner ): drmaa.JobState.FAILED: 'job finished, but failed', } - self.ds = drmaa.Session() - self.ds.initialize() + # Ensure a DRMAA session exists and is initialized + self.ds = DrmaaSessionFactory().get() self.userid = None @@ -136,17 +134,18 @@ class DRMAAJobRunner( AsynchronousJobRunner ): ajs = AsynchronousJobState( files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper, job_name=job_name ) # set up the drmaa job template - jt = self.ds.createJobTemplate() - jt.remoteCommand = ajs.job_file - jt.jobName = ajs.job_name - jt.workingDirectory = job_wrapper.working_directory - jt.outputPath = ":%s" % ajs.output_file - jt.errorPath = ":%s" % ajs.error_file + jt = dict( + remoteCommand = ajs.job_file, + jobName = ajs.job_name, + workingDirectory = job_wrapper.working_directory, + outputPath = ":%s" % ajs.output_file, + errorPath = ":%s" % ajs.error_file + ) # Avoid a jt.exitCodePath for now - it's only used when finishing. native_spec = job_destination.params.get('nativeSpecification', None) if native_spec is not None: - jt.nativeSpecification = native_spec + jt['nativeSpecification'] = native_spec # fill in the DRM's job run template script = self.get_job_file(job_wrapper, exit_code_path=ajs.exit_code_file) @@ -177,7 +176,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ): fail_msg = None while external_job_id is None and trynum < 5: try: - external_job_id = self.ds.runJob(jt) + external_job_id = self.ds.run_job(**jt) break except ( drmaa.InternalException, drmaa.DeniedByDrmException ) as e: trynum += 1 @@ -192,7 +191,6 @@ class DRMAAJobRunner( AsynchronousJobRunner ): if not fail_msg: fail_msg = DEFAULT_JOB_PUT_FAILURE_MESSAGE job_wrapper.fail( fail_msg ) - self.ds.deleteJobTemplate( jt ) return else: job_wrapper.change_ownership_for_run() @@ -203,7 +201,6 @@ class DRMAAJobRunner( AsynchronousJobRunner ): if not allow_guests: fail_msg = "User %s is not mapped to any real user, and not permitted to start jobs." % job_wrapper.user job_wrapper.fail( fail_msg ) - self.ds.deleteJobTemplate( jt ) return pwent = job_wrapper.galaxy_system_pwent log.debug( '(%s) submitting with credentials: %s [uid: %s]' % ( galaxy_id_tag, pwent[0], pwent[2] ) ) @@ -220,9 +217,6 @@ class DRMAAJobRunner( AsynchronousJobRunner ): ajs.old_state = 'new' ajs.job_destination = job_destination - # delete the job template - self.ds.deleteJobTemplate( jt ) - # Add to our 'queue' of jobs to monitor self.monitor_queue.put( ajs ) @@ -256,7 +250,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ): old_state = ajs.old_state try: assert external_job_id not in ( None, 'None' ), '(%s/%s) Invalid job id' % ( galaxy_id_tag, external_job_id ) - state = self.ds.jobStatus( external_job_id ) + state = self.ds.job_status( external_job_id ) except ( drmaa.InternalException, drmaa.InvalidJobException ) as e: if isinstance( e , drmaa.InvalidJobException ): ecn = "InvalidJobException".lower() @@ -313,7 +307,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ): assert ext_id not in ( None, 'None' ), 'External job id is None' kill_script = job.get_destination_configuration(self.app.config, "drmaa_external_killjob_script", None) if kill_script is None: - self.ds.control( ext_id, drmaa.JobControlAction.TERMINATE ) + self.ds.kill( ext_id ) else: # FIXME: hardcoded path subprocess.Popen( [ '/usr/bin/sudo', '-E', kill_script, str( ext_id ), str( self.userid ) ], shell=False ) @@ -350,16 +344,8 @@ class DRMAAJobRunner( AsynchronousJobRunner ): Path is hard-coded, but it's no worse than other path in this module. Uses Galaxy's JobID, so file is expected to be unique.""" filename = "%s/%s.jt_json" % (self.app.config.cluster_files_directory, job_wrapper.get_id_tag()) - data = {} - for attr in DRMAA_jobTemplate_attributes: - try: - data[attr] = getattr(jt, attr) - except: - pass - s = json.dumps(data) - f = open(filename, 'w+') - f.write(s) - f.close() + with open(filename, 'w+') as fp: + json.dump(jt, fp) log.debug( '(%s) Job script for external submission is: %s' % ( job_wrapper.job_id, filename ) ) return filename From a6af938ea424590fda3f0d0091dfb1ceb79f65f4 Mon Sep 17 00:00:00 2001 From: Nate Coraor Date: Wed, 6 Apr 2016 16:39:08 -0400 Subject: [PATCH 2/3] Fix linting errors --- lib/galaxy/jobs/runners/drmaa.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/lib/galaxy/jobs/runners/drmaa.py b/lib/galaxy/jobs/runners/drmaa.py index 05620d0a544..0f18144541e 100644 --- a/lib/galaxy/jobs/runners/drmaa.py +++ b/lib/galaxy/jobs/runners/drmaa.py @@ -135,11 +135,11 @@ class DRMAAJobRunner( AsynchronousJobRunner ): # set up the drmaa job template jt = dict( - remoteCommand = ajs.job_file, - jobName = ajs.job_name, - workingDirectory = job_wrapper.working_directory, - outputPath = ":%s" % ajs.output_file, - errorPath = ":%s" % ajs.error_file + remoteCommand=ajs.job_file, + jobName=ajs.job_name, + workingDirectory=job_wrapper.working_directory, + outputPath=":%s" % ajs.output_file, + errorPath=":%s" % ajs.error_file ) # Avoid a jt.exitCodePath for now - it's only used when finishing. From edd23f7c7a31e1d5113fbe82ae09324c67faf2d0 Mon Sep 17 00:00:00 2001 From: Nate Coraor Date: Thu, 7 Apr 2016 12:31:41 -0400 Subject: [PATCH 3/3] Bump pulsar-galaxy-lib version --- lib/galaxy/dependencies/pinned-requirements.txt | 2 +- lib/galaxy/dependencies/requirements.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/galaxy/dependencies/pinned-requirements.txt b/lib/galaxy/dependencies/pinned-requirements.txt index 2ee38fe4b4e..8243b7f089a 100644 --- a/lib/galaxy/dependencies/pinned-requirements.txt +++ b/lib/galaxy/dependencies/pinned-requirements.txt @@ -43,7 +43,7 @@ anyjson==0.3.3 # Pulsar requirements psutil==4.1.0 -pulsar-galaxy-lib==0.7.0.dev1 +pulsar-galaxy-lib==0.7.0.dev2 # sqlalchemy-migrate and dependencies sqlalchemy-migrate==0.10.0 diff --git a/lib/galaxy/dependencies/requirements.txt b/lib/galaxy/dependencies/requirements.txt index 23ea3cd86ae..ef49ab5dfde 100644 --- a/lib/galaxy/dependencies/requirements.txt +++ b/lib/galaxy/dependencies/requirements.txt @@ -41,7 +41,7 @@ kombu # Pulsar requirements psutil -pulsar-galaxy-lib==0.7.0.dev1 +pulsar-galaxy-lib==0.7.0.dev2 # sqlalchemy-migrate and dependencies sqlalchemy-migrate