Merged in jmchilton/galaxy-central-fork-1 (pull request #378)

Allow specification of environment variables for job destinations

job_conf.xml.sample_advanced

@@ -74,6 +74,23 @@
             <param id="Resource_List">walltime=72:00:00</param>
         </destination>
         <destination id="remote_cluster" runner="drmaa" tags="longjobs"/>
+        <destination id="java_cluster" runner="drmaa">
+          <!-- set arbitrary environment variables at runtime - like metrics
+               doesn't yet work with local or CLI runners. But should work with
+               DRMAA/SLURM, PBS, Condor, and LWR. General dependencies for tools
+               should be configured via tool_depednency_dir and package options
+               and these options should be reserved for defining cluster specific
+               options.
+          -->
+          <env id="_JAVA_OPTIONS">-Xmx=6GB</env>
+          <env id="ANOTHER_OPTION" raw="true">'5'</env> <!-- raw disables auto quoting -->
+          <env file="/mnt/java_cluster/environment_setup.sh" /> <!-- will be sourced -->
+          <env exec="module load javastuff/2.10" /> <!-- will be sourced -->
+          <!-- files to source and exec statements will be handled on remote
+               clusters. These don't need to be available on the Galaxy server
+               itself.
+          -->
+        </destination>
         <destination id="real_user_cluster" runner="drmaa">
             <!-- TODO: The real user options should maybe not be considered runner params. -->
             <param id="galaxy_external_runjob_script">scripts/drmaa_external_runner.py</param>

lib/galaxy/jobs/__init__.py

@@ -179,6 +179,7 @@ class JobConfiguration( object ):
                     job_metrics.set_destination_conf_element( id, metrics_elements[ 0 ] )
             job_destination = JobDestination(**dict(destination.items()))
             job_destination['params'] = self.__get_params(destination)
+            job_destination['env'] = self.__get_envs(destination)
             self.destinations[id] = (job_destination,)
             if job_destination.tags is not None:
                 for tag in job_destination.tags:
@@ -350,6 +351,25 @@ class JobConfiguration( object ):
             rval[param.get('id')] = param.text
         return rval
 
+    def __get_envs(self, parent):
+        """Parses any child <env> tags in to a dictionary suitable for persistence.
+
+        :param parent: Parent element in which to find child <param> tags.
+        :type parent: ``xml.etree.ElementTree.Element``
+
+        :returns: dict
+        """
+        rval = []
+        for param in parent.findall('env'):
+            rval.append( dict(
+                name=param.get('id'),
+                file=param.get('file'),
+                execute=param.get('exec'),
+                value=param.text,
+                raw=util.asbool(param.get('raw', 'false'))
+            ) )
+        return rval
+
     @property
     def default_job_tool_configuration(self):
         """The default JobToolConfiguration, used if a tool does not have an explicit defintion in the configuration.  It consists of a reference to the default handler and default destination.

lib/galaxy/jobs/runners/__init__.py

@@ -17,6 +17,7 @@ from galaxy import model
 from galaxy.util import DATABASE_MAX_STRING_SIZE, shrink_stream_by_size
 from galaxy.util import in_directory
 from galaxy.jobs.runners.util.job_script import job_script
+from galaxy.jobs.runners.util.env import env_to_statement
 
 log = logging.getLogger( __name__ )
 
@@ -253,10 +254,17 @@ class BaseJobRunner( object ):
     def get_job_file(self, job_wrapper, **kwds):
         job_metrics = job_wrapper.app.job_metrics
         job_instrumenter = job_metrics.job_instrumenters[ job_wrapper.job_destination.id ]
+
+        env_setup_commands = kwds.get( 'env_setup_commands', [] )
+        env_setup_commands.append( job_wrapper.get_env_setup_clause() or '' )
+        destination = job_wrapper.job_destination or {}
+        envs = destination.get( "env", [] )
+        for env in envs:
+            env_setup_commands.append( env_to_statement( env ) )
         options = dict(
             job_instrumenter=job_instrumenter,
             galaxy_lib=job_wrapper.galaxy_lib_dir,
-            env_setup_commands=job_wrapper.get_env_setup_clause(),
+            env_setup_commands=env_setup_commands,
             working_directory=os.path.abspath( job_wrapper.working_directory ),
             command=job_wrapper.runner_command_line,
         )

lib/galaxy/jobs/runners/lwr.py

@@ -134,6 +134,7 @@ class LwrJobRunner( AsynchronousJobRunner ):
                 tool=job_wrapper.tool,
                 config_files=job_wrapper.extra_filenames,
                 requirements=requirements,
+                env=client.env,
                 rewrite_paths=rewrite_paths,
                 arbitrary_files=unstructured_path_rewrites,
             )
@@ -222,14 +223,15 @@ class LwrJobRunner( AsynchronousJobRunner ):
         for key, value in params.iteritems():
             if value:
                 params[key] = model.User.expand_user_properties( job_wrapper.get_job().user, value )
-        return self.get_client( params, job_id )
+        env = getattr( job_wrapper.job_destination, "env", [] )
+        return self.get_client( params, job_id, env )
 
     def get_client_from_state(self, job_state):
         job_destination_params = job_state.job_destination.params
         job_id = job_state.job_id
         return self.get_client( job_destination_params, job_id )
 
-    def get_client( self, job_destination_params, job_id ):
+    def get_client( self, job_destination_params, job_id, env=[] ):
         # Cannot use url_for outside of web thread.
         #files_endpoint = url_for( controller="job_files", job_id=encoded_job_id )
 
@@ -243,6 +245,7 @@ class LwrJobRunner( AsynchronousJobRunner ):
         get_client_kwds = dict(
             job_id=str( job_id ),
             files_endpoint=files_endpoint,
+            env=env
         )
         return self.client_manager.get_client( job_destination_params, **get_client_kwds )
 

lib/galaxy/jobs/runners/lwr_client/client.py

@@ -41,6 +41,7 @@ class BaseJobClient(object):
             job_directory = None
         self.env = destination_params.get( "env", [] )
         self.files_endpoint = destination_params.get("files_endpoint", None)
+        self.env = destination_params.get("env", [])
         self.job_directory = job_directory
 
         self.default_file_action = self.destination_params.get("default_file_action", "transfer")

lib/galaxy/jobs/runners/pbs.py

@@ -296,7 +296,7 @@ class PBSJobRunner( AsynchronousJobRunner ):
         else:
             stage_commands = ''
 
-        env_setup_commands = '%s\n%s' % (stage_commands, job_wrapper.get_env_setup_clause())
+        env_setup_commands = [ stage_commands ]
         script = self.get_job_file(job_wrapper, exit_code_path=ecfile, env_setup_commands=env_setup_commands)
         job_file = "%s/%s.sh" % (self.app.config.cluster_files_directory, job_wrapper.job_id)
         fh = file(job_file, "w")

lib/galaxy/jobs/runners/util/env.py

@@ -0,0 +1,40 @@
+
+RAW_VALUE_BY_DEFAULT = False
+
+
+def env_to_statement(env):
+    ''' Return the abstraction description of an environment variable definition
+    into a statement for shell script.
+
+    >>> env_to_statement(dict(name='X', value='Y'))
+    'X="Y"; export X'
+    >>> env_to_statement(dict(name='X', value='Y', raw=True))
+    'X=Y; export X'
+    >>> env_to_statement(dict(name='X', value='"A","B","C"'))
+    'X="\\\\"A\\\\",\\\\"B\\\\",\\\\"C\\\\""; export X'
+    >>> env_to_statement(dict(file="Y"))
+    '. "Y"'
+    >>> env_to_statement(dict(file="'RAW $FILE'", raw=True))
+    ". 'RAW $FILE'"
+    >>> # Source file takes precedence
+    >>> env_to_statement(dict(name='X', value='"A","B","C"', file="S"))
+    '. "S"'
+    >>> env_to_statement(dict(execute="module load java/1.5.1"))
+    'module load java/1.5.1'
+    '''
+    source_file = env.get('file', None)
+    if source_file:
+        return '. %s' % __escape(source_file, env)
+    execute = env.get('execute', None)
+    if execute:
+        return execute
+    name = env['name']
+    value = __escape(env['value'], env)
+    return '%s=%s; export %s' % (name, value, name)
+
+
+def __escape(value, env):
+    raw = env.get('raw', RAW_VALUE_BY_DEFAULT)
+    if not raw:
+        value = '"' + value.replace('"', '\\"') + '"'
+    return value

lib/galaxy/jobs/runners/util/job_script/__init__.py

@@ -16,7 +16,7 @@ REQUIRED_TEMPLATE_PARAMS = ['working_directory', 'command', 'exit_code_path']
 OPTIONAL_TEMPLATE_PARAMS = {
     'galaxy_lib': None,
     'headers': '',
-    'env_setup_commands': '',
+    'env_setup_commands': [],
     'slots_statement': SLOTS_STATEMENT_CLUSTER_DEFAULT,
     'instrument_pre_commands': '',
     'instrument_post_commands': '',
@@ -51,13 +51,15 @@ def job_script(template=DEFAULT_JOB_FILE_TEMPLATE, **kwds):
         raise Exception("Failed to create job_script, a required parameter is missing.")
     job_instrumenter = kwds.get("job_instrumenter", None)
     if job_instrumenter:
-        del kwds[ "job_instrumenter" ]
+        del kwds["job_instrumenter"]
         working_directory = kwds["working_directory"]
         kwds["instrument_pre_commands"] = job_instrumenter.pre_execute_commands(working_directory) or ''
         kwds["instrument_post_commands"] = job_instrumenter.post_execute_commands(working_directory) or ''
 
     template_params = OPTIONAL_TEMPLATE_PARAMS.copy()
     template_params.update(**kwds)
+    env_setup_commands_str = "\n".join(template_params["env_setup_commands"])
+    template_params["env_setup_commands"] = env_setup_commands_str
     if not isinstance(template, Template):
         template = Template(template)
     return template.safe_substitute(template_params)

test/unit/jobs/test_job_configuration.py

@@ -111,6 +111,20 @@ class JobConfXmlParserTestCase( unittest.TestCase ):
         assert limits.concurrent_jobs[ "longjobs" ] == 1
         assert limits.walltime_delta == datetime.timedelta( 0, 0, 0, 0, 0, 24 )
 
+    def test_env_parsing( self ):
+        self.__with_advanced_config()
+        env_dest = self.job_config.destinations[ "java_cluster" ][ 0 ]
+        assert len( env_dest.env ) == 4, len( env_dest.env )
+        assert env_dest.env[ 0 ][ "name" ] == "_JAVA_OPTIONS"
+        assert env_dest.env[ 0 ][ "value" ] == '-Xmx=6GB'
+
+        assert env_dest.env[ 1 ][ "name" ] == "ANOTHER_OPTION"
+        assert env_dest.env[ 1 ][ "raw" ] is True
+
+        assert env_dest.env[ 2 ][ "file" ] == "/mnt/java_cluster/environment_setup.sh"
+
+        assert env_dest.env[ 3 ][ "execute" ] == "module load javastuff/2.10"
+
     # TODO: Add job metrics parsing test.
 
     @property