opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2102 from natefoo/global-drmaa-session

Browse Source 
Convert the DRMAA runner to use Pulsar's DRMAA session wrapper
This commit is contained in:
John Chilton
2016-04-08 09:47:24 -04:00
parent 87ff0e38fa edd23f7c7a
commit ef604a75e5
3 changed files with 18 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/dependencies/pinned-requirements.txt
+1 -1
View File
@@ -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
lib/galaxy/dependencies/requirements.txt
+1 -1
View File
@@ -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
lib/galaxy/jobs/runners/drmaa.py
+16 -30
View File
@@ -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