June 2014 - galaxy-commits - lists.galaxyproject.org

commit/galaxy-central: jmchilton: Make galaxy.util.in_directory slightly more generic to match Pulsar's variant.
by commits-noreply＠bitbucket.org 23 Jun '14

23 Jun '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/fbf4b5ab17a9/ Changeset: fbf4b5ab17a9 User: jmchilton Date: 2014-06-24 04:55:11 Summary: Make galaxy.util.in_directory slightly more generic to match Pulsar's variant. Fixes failing unit test in just added code. Affected #: 1 file diff -r 566388d623749dad4259ea04cf44a5d858a56671 -r fbf4b5ab17a90ec7b0deac658ead6a99b9a6ca76 lib/galaxy/util/__init__.py --- a/lib/galaxy/util/__init__.py +++ b/lib/galaxy/util/__init__.py @@ -417,17 +417,16 @@ return slug_base -def in_directory( file, directory ): +def in_directory( file, directory, local_path_module=os.path ): """ Return true, if the common prefix of both is equal to directory e.g. /a/b/c/d.rst and directory is /a/b, the common prefix is /a/b """ # Make both absolute. - directory = os.path.abspath( directory ) - file = os.path.abspath( file ) - - return os.path.commonprefix( [ file, directory ] ) == directory + directory = local_path_module.abspath(directory) + file = local_path_module.abspath(file) + return local_path_module.commonprefix([file, directory]) == directory def merge_sorted_iterables( operator, *iterables ): Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: jmchilton: Merged in jmchilton/galaxy-central-fork-1 (pull request #422)
by commits-noreply＠bitbucket.org 23 Jun '14

23 Jun '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/566388d62374/ Changeset: 566388d62374 User: jmchilton Date: 2014-06-24 04:27:51 Summary: Merged in jmchilton/galaxy-central-fork-1 (pull request #422) Implement Pulsar job runners. Affected #: 32 files diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 job_conf.xml.sample_advanced --- a/job_conf.xml.sample_advanced +++ b/job_conf.xml.sample_advanced @@ -19,27 +19,30 @@ <param id="drmaa_library_path">/sge/lib/libdrmaa.so</param></plugin> - <plugin id="lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner"> -  -  + <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" /> + <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" /> + <plugin id="slurm" type="runner" load="galaxy.jobs.runners.slurm:SlurmJobRunner" /> +  + <plugin id="pulsar_rest" type="runner" load="galaxy.jobs.runners.pulsar:PulsarRESTJobRunner"> +  -  - </plugin> - <plugin id="amqp_lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner"> - <param id="url">amqp://guest:guest@localhost:5672//</param> -  + <plugin id="pulsar_mq" type="runner" load="galaxy.jobs.runners.pulsar:PulsarMQJobRunner"> +  + <param id="amqp_url">amqp://guest:guest@localhost:5672//</param> + <param id="galaxy_url">http://localhost:8080</param> -  + </plugin> - <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" /> - <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" /> - <plugin id="slurm" type="runner" load="galaxy.jobs.runners.slurm:SlurmJobRunner" /> + <plugin id="pulsar_legacy" type="runner" load="galaxy.jobs.runners.pulsar:PulsarLegacyJobRunner"> +  + </plugin></plugins><handlers default="handlers"> @@ -213,87 +224,71 @@ <param id="function">foo</param></destination> - <destination id="secure_lwr" runner="lwr"> - <param id="url">https://windowshost.examle.com:8913/</param> -  + <destination id="secure_pulsar_rest_dest" runner="pulsar_rest"> + <param id="url">https://examle.com:8913/</param> + <param id="private_token">123456789changeme</param> -  -  +  -  -  -  +  -  -  -  +  +  + </destination> - <destination id="amqp_lwr_dest" runner="amqp_lwr" > -  -  - <param id="jobs_directory">/path/to/remote/lwr/lwr_staging/</param> -  - <param id="default_file_action">remote_transfer</param></destination><destination id="ssh_torque" runner="cli"><param id="shell_plugin">SecureShell</param> diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/pulsar.py --- /dev/null +++ b/lib/galaxy/jobs/runners/pulsar.py @@ -0,0 +1,707 @@ +from __future__ import absolute_import # Need to import pulsar_client absolutely. + +import logging + +from galaxy import model +from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner +from galaxy.jobs import ComputeEnvironment +from galaxy.jobs import JobDestination +from galaxy.jobs.command_factory import build_command +from galaxy.tools.deps import dependencies +from galaxy.util import string_as_bool_or_none +from galaxy.util.bunch import Bunch +from galaxy.util import specs + +import errno +from time import sleep +import os + +from pulsar.client import build_client_manager +from pulsar.client import url_to_destination_params +from pulsar.client import finish_job as pulsar_finish_job +from pulsar.client import submit_job as pulsar_submit_job +from pulsar.client import ClientJobDescription +from pulsar.client import PulsarOutputs +from pulsar.client import ClientOutputs +from pulsar.client import PathMapper + +log = logging.getLogger( __name__ ) + +__all__ = [ 'PulsarLegacyJobRunner', 'PulsarRESTJobRunner', 'PulsarMQJobRunner' ] + +NO_REMOTE_GALAXY_FOR_METADATA_MESSAGE = "Pulsar misconfiguration - Pulsar client configured to set metadata remotely, but remote Pulsar isn't properly configured with a galaxy_home directory." +NO_REMOTE_DATATYPES_CONFIG = "Pulsar client is configured to use remote datatypes configuration when setting metadata externally, but Pulsar is not configured with this information. Defaulting to datatypes_conf.xml." +GENERIC_REMOTE_ERROR = "Failed to communicate with remote job server." + +# Is there a good way to infer some default for this? Can only use +# url_for from web threads. https://gist.github.com/jmchilton/9098762 +DEFAULT_GALAXY_URL = "http://localhost:8080" + +PULSAR_PARAM_SPECS = dict( + transport=dict( + map=specs.to_str_or_none, + valid=specs.is_in("urllib", "curl", None), + default=None + ), + cache=dict( + map=specs.to_bool_or_none, + default=None, + ), + amqp_url=dict( + map=specs.to_str_or_none, + default=None, + ), + galaxy_url=dict( + map=specs.to_str_or_none, + default=DEFAULT_GALAXY_URL, + ), + manager=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_consumer_timeout=dict( + map=lambda val: None if val == "None" else float(val), + default=None, + ), + amqp_connect_ssl_ca_certs=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_keyfile=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_certfile=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_cert_reqs=dict( + map=specs.to_str_or_none, + default=None, + ), + # http://kombu.readthedocs.org/en/latest/reference/kombu.html#kombu.Producer.… + amqp_publish_retry=dict( + map=specs.to_bool, + default=False, + ), + amqp_publish_priority=dict( + map=int, + valid=lambda x: 0 <= x and x <= 9, + default=0, + ), + # http://kombu.readthedocs.org/en/latest/reference/kombu.html#kombu.Exchange.… + amqp_publish_delivery_mode=dict( + map=str, + valid=specs.is_in("transient", "persistent"), + default="persistent", + ), + amqp_publish_retry_max_retries=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_start=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_step=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_max=dict( + map=int, + default=None, + ), +) + + +PARAMETER_SPECIFICATION_REQUIRED = object() +PARAMETER_SPECIFICATION_IGNORED = object() + + +class PulsarJobRunner( AsynchronousJobRunner ): + """ + Pulsar Job Runner + """ + runner_name = "PulsarJobRunner" + + def __init__( self, app, nworkers, **kwds ): + """Start the job runner """ + super( PulsarJobRunner, self ).__init__( app, nworkers, runner_param_specs=PULSAR_PARAM_SPECS, **kwds ) + self._init_worker_threads() + galaxy_url = self.runner_params.galaxy_url + if galaxy_url: + galaxy_url = galaxy_url.rstrip("/") + self.galaxy_url = galaxy_url + self.__init_client_manager() + self._monitor() + + def _monitor( self ): + # Extension point allow MQ variant to setup callback instead + self._init_monitor_thread() + + def __init_client_manager( self ): + client_manager_kwargs = {} + for kwd in 'manager', 'cache', 'transport': + client_manager_kwargs[ kwd ] = self.runner_params[ kwd ] + for kwd in self.runner_params.keys(): + if kwd.startswith( 'amqp_' ): + client_manager_kwargs[ kwd ] = self.runner_params[ kwd ] + self.client_manager = build_client_manager(**client_manager_kwargs) + + def url_to_destination( self, url ): + """Convert a legacy URL to a job destination""" + return JobDestination( runner="pulsar", params=url_to_destination_params( url ) ) + + def check_watched_item(self, job_state): + try: + client = self.get_client_from_state(job_state) + status = client.get_status() + except Exception: + # An orphaned job was put into the queue at app startup, so remote server went down + # either way we are done I guess. + self.mark_as_finished(job_state) + return None + job_state = self._update_job_state_for_status(job_state, status) + return job_state + + def _update_job_state_for_status(self, job_state, pulsar_status): + if pulsar_status == "complete": + self.mark_as_finished(job_state) + return None + if pulsar_status == "failed": + self.fail_job(job_state) + return None + if pulsar_status == "running" and not job_state.running: + job_state.running = True + job_state.job_wrapper.change_state( model.Job.states.RUNNING ) + return job_state + + def queue_job(self, job_wrapper): + job_destination = job_wrapper.job_destination + self._populate_parameter_defaults( job_destination ) + + command_line, client, remote_job_config, compute_environment = self.__prepare_job( job_wrapper, job_destination ) + + if not command_line: + return + + try: + dependencies_description = PulsarJobRunner.__dependencies_description( client, job_wrapper ) + rewrite_paths = not PulsarJobRunner.__rewrite_parameters( client ) + unstructured_path_rewrites = {} + if compute_environment: + unstructured_path_rewrites = compute_environment.unstructured_path_rewrites + + client_job_description = ClientJobDescription( + command_line=command_line, + input_files=self.get_input_files(job_wrapper), + client_outputs=self.__client_outputs(client, job_wrapper), + working_directory=job_wrapper.working_directory, + tool=job_wrapper.tool, + config_files=job_wrapper.extra_filenames, + dependencies_description=dependencies_description, + env=client.env, + rewrite_paths=rewrite_paths, + arbitrary_files=unstructured_path_rewrites, + ) + job_id = pulsar_submit_job(client, client_job_description, remote_job_config) + log.info("Pulsar job submitted with job_id %s" % job_id) + job_wrapper.set_job_destination( job_destination, job_id ) + job_wrapper.change_state( model.Job.states.QUEUED ) + except Exception: + job_wrapper.fail( "failure running job", exception=True ) + log.exception("failure running job %d" % job_wrapper.job_id) + return + + pulsar_job_state = AsynchronousJobState() + pulsar_job_state.job_wrapper = job_wrapper + pulsar_job_state.job_id = job_id + pulsar_job_state.old_state = True + pulsar_job_state.running = False + pulsar_job_state.job_destination = job_destination + self.monitor_job(pulsar_job_state) + + def __prepare_job(self, job_wrapper, job_destination): + """ Build command-line and Pulsar client for this job. """ + command_line = None + client = None + remote_job_config = None + compute_environment = None + try: + client = self.get_client_from_wrapper(job_wrapper) + tool = job_wrapper.tool + remote_job_config = client.setup(tool.id, tool.version) + rewrite_parameters = PulsarJobRunner.__rewrite_parameters( client ) + prepare_kwds = {} + if rewrite_parameters: + compute_environment = PulsarComputeEnvironment( client, job_wrapper, remote_job_config ) + prepare_kwds[ 'compute_environment' ] = compute_environment + job_wrapper.prepare( **prepare_kwds ) + self.__prepare_input_files_locally(job_wrapper) + remote_metadata = PulsarJobRunner.__remote_metadata( client ) + dependency_resolution = PulsarJobRunner.__dependency_resolution( client ) + metadata_kwds = self.__build_metadata_configuration(client, job_wrapper, remote_metadata, remote_job_config) + remote_command_params = dict( + working_directory=remote_job_config['working_directory'], + metadata_kwds=metadata_kwds, + dependency_resolution=dependency_resolution, + ) + remote_working_directory = remote_job_config['working_directory'] + # TODO: Following defs work for Pulsar, always worked for Pulsar but should be + # calculated at some other level. + remote_job_directory = os.path.abspath(os.path.join(remote_working_directory, os.path.pardir)) + remote_tool_directory = os.path.abspath(os.path.join(remote_job_directory, "tool_files")) + container = self._find_container( + job_wrapper, + compute_working_directory=remote_working_directory, + compute_tool_directory=remote_tool_directory, + compute_job_directory=remote_job_directory, + ) + command_line = build_command( + self, + job_wrapper=job_wrapper, + container=container, + include_metadata=remote_metadata, + include_work_dir_outputs=False, + remote_command_params=remote_command_params, + ) + except Exception: + job_wrapper.fail( "failure preparing job", exception=True ) + log.exception("failure running job %d" % job_wrapper.job_id) + + # If we were able to get a command line, run the job + if not command_line: + job_wrapper.finish( '', '' ) + + return command_line, client, remote_job_config, compute_environment + + def __prepare_input_files_locally(self, job_wrapper): + """Run task splitting commands locally.""" + prepare_input_files_cmds = getattr(job_wrapper, 'prepare_input_files_cmds', None) + if prepare_input_files_cmds is not None: + for cmd in prepare_input_files_cmds: # run the commands to stage the input files + if 0 != os.system(cmd): + raise Exception('Error running file staging command: %s' % cmd) + job_wrapper.prepare_input_files_cmds = None # prevent them from being used in-line + + def _populate_parameter_defaults( self, job_destination ): + updated = False + params = job_destination.params + for key, value in self.destination_defaults.iteritems(): + if key in params: + if value is PARAMETER_SPECIFICATION_IGNORED: + log.warn( "Pulsar runner in selected configuration ignores parameter %s" % key ) + continue + #if self.runner_params.get( key, None ): + # # Let plugin define defaults for some parameters - + # # for instance that way jobs_directory can be + # # configured next to AMQP url (where it belongs). + # params[ key ] = self.runner_params[ key ] + # continue + + if not value: + continue + + if value is PARAMETER_SPECIFICATION_REQUIRED: + raise Exception( "Pulsar destination does not define required parameter %s" % key ) + elif value is not PARAMETER_SPECIFICATION_IGNORED: + params[ key ] = value + updated = True + return updated + + def get_output_files(self, job_wrapper): + output_paths = job_wrapper.get_output_fnames() + return [ str( o ) for o in output_paths ] # Force job_path from DatasetPath objects. + + def get_input_files(self, job_wrapper): + input_paths = job_wrapper.get_input_paths() + return [ str( i ) for i in input_paths ] # Force job_path from DatasetPath objects. + + def get_client_from_wrapper(self, job_wrapper): + job_id = job_wrapper.job_id + if hasattr(job_wrapper, 'task_id'): + job_id = "%s_%s" % (job_id, job_wrapper.task_id) + params = job_wrapper.job_destination.params.copy() + for key, value in params.iteritems(): + if value: + params[key] = model.User.expand_user_properties( job_wrapper.get_job().user, value ) + + 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, env=[] ): + # Cannot use url_for outside of web thread. + #files_endpoint = url_for( controller="job_files", job_id=encoded_job_id ) + + encoded_job_id = self.app.security.encode_id(job_id) + job_key = self.app.security.encode_id( job_id, kind="jobs_files" ) + files_endpoint = "%s/api/jobs/%s/files?job_key=%s" % ( + self.galaxy_url, + encoded_job_id, + job_key + ) + 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 ) + + def finish_job( self, job_state ): + stderr = stdout = '' + job_wrapper = job_state.job_wrapper + try: + client = self.get_client_from_state(job_state) + run_results = client.full_status() + remote_working_directory = run_results.get("working_directory", None) + stdout = run_results.get('stdout', '') + stderr = run_results.get('stderr', '') + exit_code = run_results.get('returncode', None) + pulsar_outputs = PulsarOutputs.from_status_response(run_results) + # Use Pulsar client code to transfer/copy files back + # and cleanup job if needed. + completed_normally = \ + job_wrapper.get_state() not in [ model.Job.states.ERROR, model.Job.states.DELETED ] + cleanup_job = self.app.config.cleanup_job + client_outputs = self.__client_outputs(client, job_wrapper) + finish_args = dict( client=client, + job_completed_normally=completed_normally, + cleanup_job=cleanup_job, + client_outputs=client_outputs, + pulsar_outputs=pulsar_outputs ) + failed = pulsar_finish_job( **finish_args ) + if failed: + job_wrapper.fail("Failed to find or download one or more job outputs from remote server.", exception=True) + except Exception: + message = GENERIC_REMOTE_ERROR + job_wrapper.fail( message, exception=True ) + log.exception("failure finishing job %d" % job_wrapper.job_id) + return + if not PulsarJobRunner.__remote_metadata( client ): + self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) + # Finish the job + try: + job_wrapper.finish( + stdout, + stderr, + exit_code, + remote_working_directory=remote_working_directory + ) + except Exception: + log.exception("Job wrapper finish method failed") + job_wrapper.fail("Unable to finish job", exception=True) + + def fail_job( self, job_state ): + """ + Seperated out so we can use the worker threads for it. + """ + self.stop_job( self.sa_session.query( self.app.model.Job ).get( job_state.job_wrapper.job_id ) ) + job_state.job_wrapper.fail( getattr( job_state, "fail_message", GENERIC_REMOTE_ERROR ) ) + + def check_pid( self, pid ): + try: + os.kill( pid, 0 ) + return True + except OSError, e: + if e.errno == errno.ESRCH: + log.debug( "check_pid(): PID %d is dead" % pid ) + else: + log.warning( "check_pid(): Got errno %s when attempting to check PID %d: %s" % ( errno.errorcode[e.errno], pid, e.strerror ) ) + return False + + def stop_job( self, job ): + #if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished + job_ext_output_metadata = job.get_external_output_metadata() + if job_ext_output_metadata: + pid = job_ext_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them + if pid in [ None, '' ]: + log.warning( "stop_job(): %s: no PID in database for job, unable to stop" % job.id ) + return + pid = int( pid ) + if not self.check_pid( pid ): + log.warning( "stop_job(): %s: PID %d was already dead or can't be signaled" % ( job.id, pid ) ) + return + for sig in [ 15, 9 ]: + try: + os.killpg( pid, sig ) + except OSError, e: + log.warning( "stop_job(): %s: Got errno %s when attempting to signal %d to PID %d: %s" % ( job.id, errno.errorcode[e.errno], sig, pid, e.strerror ) ) + return # give up + sleep( 2 ) + if not self.check_pid( pid ): + log.debug( "stop_job(): %s: PID %d successfully killed with signal %d" % ( job.id, pid, sig ) ) + return + else: + log.warning( "stop_job(): %s: PID %d refuses to die after signaling TERM/KILL" % ( job.id, pid ) ) + else: + # Remote kill + pulsar_url = job.job_runner_name + job_id = job.job_runner_external_id + log.debug("Attempt remote Pulsar kill of job with url %s and id %s" % (pulsar_url, job_id)) + client = self.get_client(job.destination_params, job_id) + client.kill() + + def recover( self, job, job_wrapper ): + """Recovers jobs stuck in the queued/running state when Galaxy started""" + job_state = self._job_state( job, job_wrapper ) + job_wrapper.command_line = job.get_command_line() + state = job.get_state() + if state in [model.Job.states.RUNNING, model.Job.states.QUEUED]: + log.debug( "(Pulsar/%s) is still in running state, adding to the Pulsar queue" % ( job.get_id()) ) + job_state.old_state = True + job_state.running = state == model.Job.states.RUNNING + self.monitor_queue.put( job_state ) + + def shutdown( self ): + super( PulsarJobRunner, self ).shutdown() + self.client_manager.shutdown() + + def _job_state( self, job, job_wrapper ): + job_state = AsynchronousJobState() + # TODO: Determine why this is set when using normal message queue updates + # but not CLI submitted MQ updates... + raw_job_id = job.get_job_runner_external_id() or job_wrapper.job_id + job_state.job_id = str( raw_job_id ) + job_state.runner_url = job_wrapper.get_job_runner_url() + job_state.job_destination = job_wrapper.job_destination + job_state.job_wrapper = job_wrapper + return job_state + + def __client_outputs( self, client, job_wrapper ): + work_dir_outputs = self.get_work_dir_outputs( job_wrapper ) + output_files = self.get_output_files( job_wrapper ) + client_outputs = ClientOutputs( + working_directory=job_wrapper.working_directory, + work_dir_outputs=work_dir_outputs, + output_files=output_files, + version_file=job_wrapper.get_version_string_path(), + ) + return client_outputs + + @staticmethod + def __dependencies_description( pulsar_client, job_wrapper ): + dependency_resolution = PulsarJobRunner.__dependency_resolution( pulsar_client ) + remote_dependency_resolution = dependency_resolution == "remote" + if not remote_dependency_resolution: + return None + requirements = job_wrapper.tool.requirements or [] + installed_tool_dependencies = job_wrapper.tool.installed_tool_dependencies or [] + return dependencies.DependenciesDescription( + requirements=requirements, + installed_tool_dependencies=installed_tool_dependencies, + ) + + @staticmethod + def __dependency_resolution( pulsar_client ): + dependency_resolution = pulsar_client.destination_params.get( "dependency_resolution", "local" ) + if dependency_resolution not in ["none", "local", "remote"]: + raise Exception("Unknown dependency_resolution value encountered %s" % dependency_resolution) + return dependency_resolution + + @staticmethod + def __remote_metadata( pulsar_client ): + remote_metadata = string_as_bool_or_none( pulsar_client.destination_params.get( "remote_metadata", False ) ) + return remote_metadata + + @staticmethod + def __use_remote_datatypes_conf( pulsar_client ): + """ When setting remote metadata, use integrated datatypes from this + Galaxy instance or use the datatypes config configured via the remote + Pulsar. + + Both options are broken in different ways for same reason - datatypes + may not match. One can push the local datatypes config to the remote + server - but there is no guarentee these datatypes will be defined + there. Alternatively, one can use the remote datatype config - but + there is no guarentee that it will contain all the datatypes available + to this Galaxy. + """ + use_remote_datatypes = string_as_bool_or_none( pulsar_client.destination_params.get( "use_remote_datatypes", False ) ) + return use_remote_datatypes + + @staticmethod + def __rewrite_parameters( pulsar_client ): + return string_as_bool_or_none( pulsar_client.destination_params.get( "rewrite_parameters", False ) ) or False + + def __build_metadata_configuration(self, client, job_wrapper, remote_metadata, remote_job_config): + metadata_kwds = {} + if remote_metadata: + remote_system_properties = remote_job_config.get("system_properties", {}) + remote_galaxy_home = remote_system_properties.get("galaxy_home", None) + if not remote_galaxy_home: + raise Exception(NO_REMOTE_GALAXY_FOR_METADATA_MESSAGE) + metadata_kwds['exec_dir'] = remote_galaxy_home + outputs_directory = remote_job_config['outputs_directory'] + configs_directory = remote_job_config['configs_directory'] + working_directory = remote_job_config['working_directory'] + # For metadata calculation, we need to build a list of of output + # file objects with real path indicating location on Galaxy server + # and false path indicating location on compute server. Since the + # Pulsar disables from_work_dir copying as part of the job command + # line we need to take the list of output locations on the Pulsar + # server (produced by self.get_output_files(job_wrapper)) and for + # each work_dir output substitute the effective path on the Pulsar + # server relative to the remote working directory as the + # false_path to send the metadata command generation module. + work_dir_outputs = self.get_work_dir_outputs(job_wrapper, job_working_directory=working_directory) + outputs = [Bunch(false_path=os.path.join(outputs_directory, os.path.basename(path)), real_path=path) for path in self.get_output_files(job_wrapper)] + for output in outputs: + for pulsar_workdir_path, real_path in work_dir_outputs: + if real_path == output.real_path: + output.false_path = pulsar_workdir_path + metadata_kwds['output_fnames'] = outputs + metadata_kwds['compute_tmp_dir'] = working_directory + metadata_kwds['config_root'] = remote_galaxy_home + default_config_file = os.path.join(remote_galaxy_home, 'universe_wsgi.ini') + metadata_kwds['config_file'] = remote_system_properties.get('galaxy_config_file', default_config_file) + metadata_kwds['dataset_files_path'] = remote_system_properties.get('galaxy_dataset_files_path', None) + if PulsarJobRunner.__use_remote_datatypes_conf( client ): + remote_datatypes_config = remote_system_properties.get('galaxy_datatypes_config_file', None) + if not remote_datatypes_config: + log.warn(NO_REMOTE_DATATYPES_CONFIG) + remote_datatypes_config = os.path.join(remote_galaxy_home, 'datatypes_conf.xml') + metadata_kwds['datatypes_config'] = remote_datatypes_config + else: + integrates_datatypes_config = self.app.datatypes_registry.integrated_datatypes_configs + # Ensure this file gets pushed out to the remote config dir. + job_wrapper.extra_filenames.append(integrates_datatypes_config) + + metadata_kwds['datatypes_config'] = os.path.join(configs_directory, os.path.basename(integrates_datatypes_config)) + return metadata_kwds + + +class PulsarLegacyJobRunner( PulsarJobRunner ): + destination_defaults = dict( + rewrite_parameters="false", + dependency_resolution="local", + ) + + +class PulsarMQJobRunner( PulsarJobRunner ): + destination_defaults = dict( + default_file_action="remote_transfer", + rewrite_parameters="true", + dependency_resolution="remote", + jobs_directory=PARAMETER_SPECIFICATION_REQUIRED, + url=PARAMETER_SPECIFICATION_IGNORED, + private_token=PARAMETER_SPECIFICATION_IGNORED + ) + + def _monitor( self ): + # This is a message queue driven runner, don't monitor + # just setup required callback. + self.client_manager.ensure_has_status_update_callback(self.__async_update) + + def __async_update( self, full_status ): + job_id = None + try: + job_id = full_status[ "job_id" ] + job, job_wrapper = self.app.job_manager.job_handler.job_queue.job_pair_for_id( job_id ) + job_state = self._job_state( job, job_wrapper ) + self._update_job_state_for_status(job_state, full_status[ "status" ] ) + except Exception: + log.exception( "Failed to update Pulsar job status for job_id %s" % job_id ) + raise + # Nothing else to do? - Attempt to fail the job? + + +class PulsarRESTJobRunner( PulsarJobRunner ): + destination_defaults = dict( + default_file_action="transfer", + rewrite_parameters="true", + dependency_resolution="remote", + url=PARAMETER_SPECIFICATION_REQUIRED, + ) + + +class PulsarComputeEnvironment( ComputeEnvironment ): + + def __init__( self, pulsar_client, job_wrapper, remote_job_config ): + self.pulsar_client = pulsar_client + self.job_wrapper = job_wrapper + self.local_path_config = job_wrapper.default_compute_environment() + self.unstructured_path_rewrites = {} + # job_wrapper.prepare is going to expunge the job backing the following + # computations, so precalculate these paths. + self._wrapper_input_paths = self.local_path_config.input_paths() + self._wrapper_output_paths = self.local_path_config.output_paths() + self.path_mapper = PathMapper(pulsar_client, remote_job_config, self.local_path_config.working_directory()) + self._config_directory = remote_job_config[ "configs_directory" ] + self._working_directory = remote_job_config[ "working_directory" ] + self._sep = remote_job_config[ "system_properties" ][ "separator" ] + self._tool_dir = remote_job_config[ "tools_directory" ] + version_path = self.local_path_config.version_path() + new_version_path = self.path_mapper.remote_version_path_rewrite(version_path) + if new_version_path: + version_path = new_version_path + self._version_path = version_path + + def output_paths( self ): + local_output_paths = self._wrapper_output_paths + + results = [] + for local_output_path in local_output_paths: + wrapper_path = str( local_output_path ) + remote_path = self.path_mapper.remote_output_path_rewrite( wrapper_path ) + results.append( self._dataset_path( local_output_path, remote_path ) ) + return results + + def input_paths( self ): + local_input_paths = self._wrapper_input_paths + + results = [] + for local_input_path in local_input_paths: + wrapper_path = str( local_input_path ) + # This will over-copy in some cases. For instance in the case of task + # splitting, this input will be copied even though only the work dir + # input will actually be used. + remote_path = self.path_mapper.remote_input_path_rewrite( wrapper_path ) + results.append( self._dataset_path( local_input_path, remote_path ) ) + return results + + def _dataset_path( self, local_dataset_path, remote_path ): + remote_extra_files_path = None + if remote_path: + remote_extra_files_path = "%s_files" % remote_path[ 0:-len( ".dat" ) ] + return local_dataset_path.with_path_for_job( remote_path, remote_extra_files_path ) + + def working_directory( self ): + return self._working_directory + + def config_directory( self ): + return self._config_directory + + def new_file_path( self ): + return self.working_directory() # Problems with doing this? + + def sep( self ): + return self._sep + + def version_path( self ): + return self._version_path + + def rewriter( self, parameter_value ): + unstructured_path_rewrites = self.unstructured_path_rewrites + if parameter_value in unstructured_path_rewrites: + # Path previously mapped, use previous mapping. + return unstructured_path_rewrites[ parameter_value ] + if parameter_value in unstructured_path_rewrites.itervalues(): + # Path is a rewritten remote path (this might never occur, + # consider dropping check...) + return parameter_value + + rewrite, new_unstructured_path_rewrites = self.path_mapper.check_for_arbitrary_rewrite( parameter_value ) + if rewrite: + unstructured_path_rewrites.update(new_unstructured_path_rewrites) + return rewrite + else: + # Did need to rewrite, use original path or value. + return parameter_value + + def unstructured_path_rewriter( self ): + return self.rewriter diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/util/__init__.py --- a/lib/galaxy/jobs/runners/util/__init__.py +++ b/lib/galaxy/jobs/runners/util/__init__.py @@ -1,7 +1,7 @@ """ This module and its submodules contains utilities for running external processes and interfacing with job managers. This module should contain -functionality shared between Galaxy and the LWR. +functionality shared between Galaxy and the Pulsar. """ from galaxy.util.bunch import Bunch diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/util/cli/factory.py --- a/lib/galaxy/jobs/runners/util/cli/factory.py +++ b/lib/galaxy/jobs/runners/util/cli/factory.py @@ -5,7 +5,7 @@ ) code_dir = 'lib' except ImportError: - from lwr.managers.util.cli import ( + from pulsar.managers.util.cli import ( CliInterface, split_params ) diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/util/cli/job/slurm.py --- a/lib/galaxy/jobs/runners/util/cli/job/slurm.py +++ b/lib/galaxy/jobs/runners/util/cli/job/slurm.py @@ -5,7 +5,7 @@ from galaxy.model import Job job_states = Job.states except ImportError: - # Not in Galaxy, map Galaxy job states to LWR ones. + # Not in Galaxy, map Galaxy job states to Pulsar ones. from galaxy.util import enum job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/util/cli/job/torque.py --- a/lib/galaxy/jobs/runners/util/cli/job/torque.py +++ b/lib/galaxy/jobs/runners/util/cli/job/torque.py @@ -7,7 +7,7 @@ from galaxy.model import Job job_states = Job.states except ImportError: - # Not in Galaxy, map Galaxy job states to LWR ones. + # Not in Galaxy, map Galaxy job states to Pulsar ones. from galaxy.util import enum job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/objectstore/__init__.py --- a/lib/galaxy/objectstore/__init__.py +++ b/lib/galaxy/objectstore/__init__.py @@ -623,9 +623,9 @@ elif store == 'irods': from .rods import IRODSObjectStore return IRODSObjectStore(config=config, config_xml=config_xml) - elif store == 'lwr': - from .lwr import LwrObjectStore - return LwrObjectStore(config=config, config_xml=config_xml) + elif store == 'pulsar': + from .pulsar import PulsarObjectStore + return PulsarObjectStore(config=config, config_xml=config_xml) else: log.error("Unrecognized object store definition: {0}".format(store)) diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/objectstore/lwr.py --- a/lib/galaxy/objectstore/lwr.py +++ /dev/null @@ -1,76 +0,0 @@ -from __future__ import absolute_import # Need to import lwr_client absolutely. -from ..objectstore import ObjectStore -try: - from galaxy.jobs.runners.lwr_client.manager import ObjectStoreClientManager -except ImportError: - from lwr.lwr_client.manager import ObjectStoreClientManager - - -class LwrObjectStore(ObjectStore): - """ - Object store implementation that delegates to a remote LWR server. - - This may be more aspirational than practical for now, it would be good to - Galaxy to a point that a handler thread could be setup that doesn't attempt - to access the disk files returned by a (this) object store - just passing - them along to the LWR unmodified. That modification - along with this - implementation and LWR job destinations would then allow Galaxy to fully - manage jobs on remote servers with completely different mount points. - - This implementation should be considered beta and may be dropped from - Galaxy at some future point or significantly modified. - """ - - def __init__(self, config, config_xml): - self.lwr_client = self.__build_lwr_client(config_xml) - - def exists(self, obj, **kwds): - return self.lwr_client.exists(**self.__build_kwds(obj, **kwds)) - - def file_ready(self, obj, **kwds): - return self.lwr_client.file_ready(**self.__build_kwds(obj, **kwds)) - - def create(self, obj, **kwds): - return self.lwr_client.create(**self.__build_kwds(obj, **kwds)) - - def empty(self, obj, **kwds): - return self.lwr_client.empty(**self.__build_kwds(obj, **kwds)) - - def size(self, obj, **kwds): - return self.lwr_client.size(**self.__build_kwds(obj, **kwds)) - - def delete(self, obj, **kwds): - return self.lwr_client.delete(**self.__build_kwds(obj, **kwds)) - - # TODO: Optimize get_data. - def get_data(self, obj, **kwds): - return self.lwr_client.get_data(**self.__build_kwds(obj, **kwds)) - - def get_filename(self, obj, **kwds): - return self.lwr_client.get_filename(**self.__build_kwds(obj, **kwds)) - - def update_from_file(self, obj, **kwds): - return self.lwr_client.update_from_file(**self.__build_kwds(obj, **kwds)) - - def get_store_usage_percent(self): - return self.lwr_client.get_store_usage_percent() - - def get_object_url(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None): - return None - - def __build_kwds(self, obj, **kwds): - kwds['object_id'] = obj.id - return kwds - pass - - def __build_lwr_client(self, config_xml): - url = config_xml.get("url") - private_token = config_xml.get("private_token", None) - transport = config_xml.get("transport", None) - manager_options = dict(transport=transport) - client_options = dict(url=url, private_token=private_token) - lwr_client = ObjectStoreClientManager(**manager_options).get_client(client_options) - return lwr_client - - def shutdown(self): - pass diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/objectstore/pulsar.py --- /dev/null +++ b/lib/galaxy/objectstore/pulsar.py @@ -0,0 +1,73 @@ +from __future__ import absolute_import # Need to import pulsar_client absolutely. +from ..objectstore import ObjectStore +from pulsar.client.manager import ObjectStoreClientManager + + +class PulsarObjectStore(ObjectStore): + """ + Object store implementation that delegates to a remote Pulsar server. + + This may be more aspirational than practical for now, it would be good to + Galaxy to a point that a handler thread could be setup that doesn't attempt + to access the disk files returned by a (this) object store - just passing + them along to the Pulsar unmodified. That modification - along with this + implementation and Pulsar job destinations would then allow Galaxy to fully + manage jobs on remote servers with completely different mount points. + + This implementation should be considered beta and may be dropped from + Galaxy at some future point or significantly modified. + """ + + def __init__(self, config, config_xml): + self.pulsar_client = self.__build_pulsar_client(config_xml) + + def exists(self, obj, **kwds): + return self.pulsar_client.exists(**self.__build_kwds(obj, **kwds)) + + def file_ready(self, obj, **kwds): + return self.pulsar_client.file_ready(**self.__build_kwds(obj, **kwds)) + + def create(self, obj, **kwds): + return self.pulsar_client.create(**self.__build_kwds(obj, **kwds)) + + def empty(self, obj, **kwds): + return self.pulsar_client.empty(**self.__build_kwds(obj, **kwds)) + + def size(self, obj, **kwds): + return self.pulsar_client.size(**self.__build_kwds(obj, **kwds)) + + def delete(self, obj, **kwds): + return self.pulsar_client.delete(**self.__build_kwds(obj, **kwds)) + + # TODO: Optimize get_data. + def get_data(self, obj, **kwds): + return self.pulsar_client.get_data(**self.__build_kwds(obj, **kwds)) + + def get_filename(self, obj, **kwds): + return self.pulsar_client.get_filename(**self.__build_kwds(obj, **kwds)) + + def update_from_file(self, obj, **kwds): + return self.pulsar_client.update_from_file(**self.__build_kwds(obj, **kwds)) + + def get_store_usage_percent(self): + return self.pulsar_client.get_store_usage_percent() + + def get_object_url(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None): + return None + + def __build_kwds(self, obj, **kwds): + kwds['object_id'] = obj.id + return kwds + pass + + def __build_pulsar_client(self, config_xml): + url = config_xml.get("url") + private_token = config_xml.get("private_token", None) + transport = config_xml.get("transport", None) + manager_options = dict(transport=transport) + client_options = dict(url=url, private_token=private_token) + pulsar_client = ObjectStoreClientManager(**manager_options).get_client(client_options) + return pulsar_client + + def shutdown(self): + pass diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/tools/deps/dependencies.py --- a/lib/galaxy/tools/deps/dependencies.py +++ b/lib/galaxy/tools/deps/dependencies.py @@ -8,7 +8,7 @@ related context required to resolve dependencies via the ToolShedPackageDependencyResolver. - This is meant to enable remote resolution of dependencies, by the LWR or + This is meant to enable remote resolution of dependencies, by the Pulsar or other potential remote execution mechanisms. """ @@ -39,7 +39,7 @@ @staticmethod def _toolshed_install_dependency_from_dict(as_dict): - # Rather than requiring full models in LWR, just use simple objects + # Rather than requiring full models in Pulsar, just use simple objects # containing only properties and associations used to resolve # dependencies for tool execution. repository_object = bunch.Bunch( diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/pulsar/client/__init__.py --- /dev/null +++ b/lib/pulsar/client/__init__.py @@ -0,0 +1,62 @@ +""" +pulsar client +====== + +This module contains logic for interfacing with an external Pulsar server. + +------------------ +Configuring Galaxy +------------------ + +Galaxy job runners are configured in Galaxy's ``job_conf.xml`` file. See ``job_conf.xml.sample_advanced`` +in your Galaxy code base or on +`Bitbucket <https://bitbucket.org/galaxy/galaxy-dist/src/tip/job_conf.xml.sample_advanc…>`_ +for information on how to configure Galaxy to interact with the Pulsar. + +Galaxy also supports an older, less rich configuration of job runners directly +in its main ``universe_wsgi.ini`` file. The following section describes how to +configure Galaxy to communicate with the Pulsar in this legacy mode. + +Legacy +------ + +A Galaxy tool can be configured to be executed remotely via Pulsar by +adding a line to the ``universe_wsgi.ini`` file under the +``galaxy:tool_runners`` section with the format:: + + <tool_id> = pulsar://http://<pulsar_host>:<pulsar_port> + +As an example, if a host named remotehost is running the Pulsar server +application on port ``8913``, then the tool with id ``test_tool`` can +be configured to run remotely on remotehost by adding the following +line to ``universe.ini``:: + + test_tool = pulsar://http://remotehost:8913 + +Remember this must be added after the ``[galaxy:tool_runners]`` header +in the ``universe.ini`` file. + + +""" + +from .staging.down import finish_job +from .staging.up import submit_job +from .staging import ClientJobDescription +from .staging import PulsarOutputs +from .staging import ClientOutputs +from .client import OutputNotFoundException +from .manager import build_client_manager +from .destination import url_to_destination_params +from .path_mapper import PathMapper + +__all__ = [ + build_client_manager, + OutputNotFoundException, + url_to_destination_params, + finish_job, + submit_job, + ClientJobDescription, + PulsarOutputs, + ClientOutputs, + PathMapper, +] diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/pulsar/client/action_mapper.py --- /dev/null +++ b/lib/pulsar/client/action_mapper.py @@ -0,0 +1,567 @@ +from json import load +from os import makedirs +from os.path import exists +from os.path import abspath +from os.path import dirname +from os.path import join +from os.path import basename +from os.path import sep +import fnmatch +from re import compile +from re import escape +import galaxy.util +from galaxy.util.bunch import Bunch +from .config_util import read_file +from .util import directory_files +from .util import unique_path_prefix +from .transport import get_file +from .transport import post_file + + +DEFAULT_MAPPED_ACTION = 'transfer' # Not really clear to me what this should be, exception? +DEFAULT_PATH_MAPPER_TYPE = 'prefix' + +STAGING_ACTION_REMOTE = "remote" +STAGING_ACTION_LOCAL = "local" +STAGING_ACTION_NONE = None +STAGING_ACTION_DEFAULT = "default" + +# Poor man's enum. +path_type = Bunch( + # Galaxy input datasets and extra files. + INPUT="input", + # Galaxy config and param files. + CONFIG="config", + # Files from tool's tool_dir (for now just wrapper if available). + TOOL="tool", + # Input work dir files - e.g. metadata files, task-split input files, etc.. + WORKDIR="workdir", + # Galaxy output datasets in their final home. + OUTPUT="output", + # Galaxy from_work_dir output paths and other files (e.g. galaxy.json) + OUTPUT_WORKDIR="output_workdir", + # Other fixed tool parameter paths (likely coming from tool data, but not + # nessecarily). Not sure this is the best name... + UNSTRUCTURED="unstructured", +) + + +ACTION_DEFAULT_PATH_TYPES = [ + path_type.INPUT, + path_type.CONFIG, + path_type.TOOL, + path_type.WORKDIR, + path_type.OUTPUT, + path_type.OUTPUT_WORKDIR, +] +ALL_PATH_TYPES = ACTION_DEFAULT_PATH_TYPES + [path_type.UNSTRUCTURED] + + +class FileActionMapper(object): + """ + Objects of this class define how paths are mapped to actions. + + >>> json_string = r'''{"paths": [ \ + {"path": "/opt/galaxy", "action": "none"}, \ + {"path": "/galaxy/data", "action": "transfer"}, \ + {"path": "/cool/bamfiles/**/*.bam", "action": "copy", "match_type": "glob"}, \ + {"path": ".*/dataset_\\\\d+.dat", "action": "copy", "match_type": "regex"} \ + ]}''' + >>> from tempfile import NamedTemporaryFile + >>> from os import unlink + >>> def mapper_for(default_action, config_contents): + ... f = NamedTemporaryFile(delete=False) + ... f.write(config_contents.encode('UTF-8')) + ... f.close() + ... mock_client = Bunch(default_file_action=default_action, action_config_path=f.name, files_endpoint=None) + ... mapper = FileActionMapper(mock_client) + ... mapper = FileActionMapper(config=mapper.to_dict()) # Serialize and deserialize it to make sure still works + ... unlink(f.name) + ... return mapper + >>> mapper = mapper_for(default_action='none', config_contents=json_string) + >>> # Test first config line above, implicit path prefix mapper + >>> action = mapper.action('/opt/galaxy/tools/filters/catWrapper.py', 'input') + >>> action.action_type == u'none' + True + >>> action.staging_needed + False + >>> # Test another (2nd) mapper, this one with a different action + >>> action = mapper.action('/galaxy/data/files/000/dataset_1.dat', 'input') + >>> action.action_type == u'transfer' + True + >>> action.staging_needed + True + >>> # Always at least copy work_dir outputs. + >>> action = mapper.action('/opt/galaxy/database/working_directory/45.sh', 'workdir') + >>> action.action_type == u'copy' + True + >>> action.staging_needed + True + >>> # Test glob mapper (matching test) + >>> mapper.action('/cool/bamfiles/projectABC/study1/patient3.bam', 'input').action_type == u'copy' + True + >>> # Test glob mapper (non-matching test) + >>> mapper.action('/cool/bamfiles/projectABC/study1/patient3.bam.bai', 'input').action_type == u'none' + True + >>> # Regex mapper test. + >>> mapper.action('/old/galaxy/data/dataset_10245.dat', 'input').action_type == u'copy' + True + >>> # Doesn't map unstructured paths by default + >>> mapper.action('/old/galaxy/data/dataset_10245.dat', 'unstructured').action_type == u'none' + True + >>> input_only_mapper = mapper_for(default_action="none", config_contents=r'''{"paths": [ \ + {"path": "/", "action": "transfer", "path_types": "input"} \ + ] }''') + >>> input_only_mapper.action('/dataset_1.dat', 'input').action_type == u'transfer' + True + >>> input_only_mapper.action('/dataset_1.dat', 'output').action_type == u'none' + True + >>> unstructured_mapper = mapper_for(default_action="none", config_contents=r'''{"paths": [ \ + {"path": "/", "action": "transfer", "path_types": "*any*"} \ + ] }''') + >>> unstructured_mapper.action('/old/galaxy/data/dataset_10245.dat', 'unstructured').action_type == u'transfer' + True + """ + + def __init__(self, client=None, config=None): + if config is None and client is None: + message = "FileActionMapper must be constructed from either a client or a config dictionary." + raise Exception(message) + if config is None: + config = self.__client_to_config(client) + self.default_action = config.get("default_action", "transfer") + self.mappers = mappers_from_dicts(config.get("paths", [])) + self.files_endpoint = config.get("files_endpoint", None) + + def action(self, path, type, mapper=None): + mapper = self.__find_mapper(path, type, mapper) + action_class = self.__action_class(path, type, mapper) + file_lister = DEFAULT_FILE_LISTER + action_kwds = {} + if mapper: + file_lister = mapper.file_lister + action_kwds = mapper.action_kwds + action = action_class(path, file_lister=file_lister, **action_kwds) + self.__process_action(action, type) + return action + + def unstructured_mappers(self): + """ Return mappers that will map 'unstructured' files (i.e. go beyond + mapping inputs, outputs, and config files). + """ + return filter(lambda m: path_type.UNSTRUCTURED in m.path_types, self.mappers) + + def to_dict(self): + return dict( + default_action=self.default_action, + files_endpoint=self.files_endpoint, + paths=map(lambda m: m.to_dict(), self.mappers) + ) + + def __client_to_config(self, client): + action_config_path = client.action_config_path + if action_config_path: + config = read_file(action_config_path) + else: + config = dict() + config["default_action"] = client.default_file_action + config["files_endpoint"] = client.files_endpoint + return config + + def __load_action_config(self, path): + config = load(open(path, 'rb')) + self.mappers = mappers_from_dicts(config.get('paths', [])) + + def __find_mapper(self, path, type, mapper=None): + if not mapper: + normalized_path = abspath(path) + for query_mapper in self.mappers: + if query_mapper.matches(normalized_path, type): + mapper = query_mapper + break + return mapper + + def __action_class(self, path, type, mapper): + action_type = self.default_action if type in ACTION_DEFAULT_PATH_TYPES else "none" + if mapper: + action_type = mapper.action_type + if type in ["workdir", "output_workdir"] and action_type == "none": + # We are changing the working_directory relative to what + # Galaxy would use, these need to be copied over. + action_type = "copy" + action_class = actions.get(action_type, None) + if action_class is None: + message_template = "Unknown action_type encountered %s while trying to map path %s" + message_args = (action_type, path) + raise Exception(message_template % message_args) + return action_class + + def __process_action(self, action, file_type): + """ Extension point to populate extra action information after an + action has been created. + """ + if action.action_type == "remote_transfer": + url_base = self.files_endpoint + if not url_base: + raise Exception("Attempted to use remote_transfer action with defining a files_endpoint") + if "?" not in url_base: + url_base = "%s?" % url_base + # TODO: URL encode path. + url = "%s&path=%s&file_type=%s" % (url_base, action.path, file_type) + action.url = url + +REQUIRED_ACTION_KWD = object() + + +class BaseAction(object): + action_spec = {} + + def __init__(self, path, file_lister=None): + self.path = path + self.file_lister = file_lister or DEFAULT_FILE_LISTER + + def unstructured_map(self, path_helper): + unstructured_map = self.file_lister.unstructured_map(self.path) + if self.staging_needed: + # To ensure uniqueness, prepend unique prefix to each name + prefix = unique_path_prefix(self.path) + for path, name in unstructured_map.iteritems(): + unstructured_map[path] = join(prefix, name) + else: + path_rewrites = {} + for path in unstructured_map: + rewrite = self.path_rewrite(path_helper, path) + if rewrite: + path_rewrites[path] = rewrite + unstructured_map = path_rewrites + return unstructured_map + + @property + def staging_needed(self): + return self.staging != STAGING_ACTION_NONE + + @property + def staging_action_local(self): + return self.staging == STAGING_ACTION_LOCAL + + +class NoneAction(BaseAction): + """ This action indicates the corresponding path does not require any + additional action. This should indicate paths that are available both on + the Pulsar client (i.e. Galaxy server) and remote Pulsar server with the same + paths. """ + action_type = "none" + staging = STAGING_ACTION_NONE + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type) + + @classmethod + def from_dict(cls, action_dict): + return NoneAction(path=action_dict["path"]) + + def path_rewrite(self, path_helper, path=None): + return None + + +class RewriteAction(BaseAction): + """ This actin indicates the Pulsar server should simply rewrite the path + to the specified file. + """ + action_spec = dict( + source_directory=REQUIRED_ACTION_KWD, + destination_directory=REQUIRED_ACTION_KWD + ) + action_type = "rewrite" + staging = STAGING_ACTION_NONE + + def __init__(self, path, file_lister=None, source_directory=None, destination_directory=None): + self.path = path + self.file_lister = file_lister or DEFAULT_FILE_LISTER + self.source_directory = source_directory + self.destination_directory = destination_directory + + def to_dict(self): + return dict( + path=self.path, + action_type=self.action_type, + source_directory=self.source_directory, + destination_directory=self.destination_directory, + ) + + @classmethod + def from_dict(cls, action_dict): + return RewriteAction( + path=action_dict["path"], + source_directory=action_dict["source_directory"], + destination_directory=action_dict["destination_directory"], + ) + + def path_rewrite(self, path_helper, path=None): + if not path: + path = self.path + new_path = path_helper.from_posix_with_new_base(self.path, self.source_directory, self.destination_directory) + return None if new_path == self.path else new_path + + +class TransferAction(BaseAction): + """ This actions indicates that the Pulsar client should initiate an HTTP + transfer of the corresponding path to the remote Pulsar server before + launching the job. """ + action_type = "transfer" + staging = STAGING_ACTION_LOCAL + + +class CopyAction(BaseAction): + """ This action indicates that the Pulsar client should execute a file system + copy of the corresponding path to the Pulsar staging directory prior to + launching the corresponding job. """ + action_type = "copy" + staging = STAGING_ACTION_LOCAL + + +class RemoteCopyAction(BaseAction): + """ This action indicates the Pulsar server should copy the file before + execution via direct file system copy. This is like a CopyAction, but + it indicates the action should occur on the Pulsar server instead of on + the client. + """ + action_type = "remote_copy" + staging = STAGING_ACTION_REMOTE + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type) + + @classmethod + def from_dict(cls, action_dict): + return RemoteCopyAction(path=action_dict["path"]) + + def write_to_path(self, path): + galaxy.util.copy_to_path(open(self.path, "rb"), path) + + def write_from_path(self, pulsar_path): + destination = self.path + parent_directory = dirname(destination) + if not exists(parent_directory): + makedirs(parent_directory) + with open(pulsar_path, "rb") as f: + galaxy.util.copy_to_path(f, destination) + + +class RemoteTransferAction(BaseAction): + """ This action indicates the Pulsar server should copy the file before + execution via direct file system copy. This is like a CopyAction, but + it indicates the action should occur on the Pulsar server instead of on + the client. + """ + action_type = "remote_transfer" + staging = STAGING_ACTION_REMOTE + + def __init__(self, path, file_lister=None, url=None): + super(RemoteTransferAction, self).__init__(path, file_lister=file_lister) + self.url = url + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type, url=self.url) + + @classmethod + def from_dict(cls, action_dict): + return RemoteTransferAction(path=action_dict["path"], url=action_dict["url"]) + + def write_to_path(self, path): + get_file(self.url, path) + + def write_from_path(self, pulsar_path): + post_file(self.url, pulsar_path) + + +class MessageAction(object): + """ Sort of pseudo action describing "files" store in memory and + transferred via message (HTTP, Python-call, MQ, etc...) + """ + action_type = "message" + staging = STAGING_ACTION_DEFAULT + + def __init__(self, contents, client=None): + self.contents = contents + self.client = client + + @property + def staging_needed(self): + return True + + @property + def staging_action_local(self): + # Ekkk, cannot be called if created through from_dict. + # Shouldn't be a problem the way it is used - but is an + # object design problem. + return self.client.prefer_local_staging + + def to_dict(self): + return dict(contents=self.contents, action_type=MessageAction.action_type) + + @classmethod + def from_dict(cls, action_dict): + return MessageAction(contents=action_dict["contents"]) + + def write_to_path(self, path): + open(path, "w").write(self.contents) + + +DICTIFIABLE_ACTION_CLASSES = [RemoteCopyAction, RemoteTransferAction, MessageAction] + + +def from_dict(action_dict): + action_type = action_dict.get("action_type", None) + target_class = None + for action_class in DICTIFIABLE_ACTION_CLASSES: + if action_type == action_class.action_type: + target_class = action_class + if not target_class: + message = "Failed to recover action from dictionary - invalid action type specified %s." % action_type + raise Exception(message) + return target_class.from_dict(action_dict) + + +class BasePathMapper(object): + + def __init__(self, config): + action_type = config.get('action', DEFAULT_MAPPED_ACTION) + action_class = actions.get(action_type, None) + action_kwds = action_class.action_spec.copy() + for key, value in action_kwds.items(): + if key in config: + action_kwds[key] = config[key] + elif value is REQUIRED_ACTION_KWD: + message_template = "action_type %s requires key word argument %s" + message = message_template % (action_type, key) + raise Exception(message) + self.action_type = action_type + self.action_kwds = action_kwds + path_types_str = config.get('path_types', "*defaults*") + path_types_str = path_types_str.replace("*defaults*", ",".join(ACTION_DEFAULT_PATH_TYPES)) + path_types_str = path_types_str.replace("*any*", ",".join(ALL_PATH_TYPES)) + self.path_types = path_types_str.split(",") + self.file_lister = FileLister(config) + + def matches(self, path, path_type): + path_type_matches = path_type in self.path_types + return path_type_matches and self._path_matches(path) + + def _extend_base_dict(self, **kwds): + base_dict = dict( + action=self.action_type, + path_types=",".join(self.path_types), + match_type=self.match_type + ) + base_dict.update(self.file_lister.to_dict()) + base_dict.update(self.action_kwds) + base_dict.update(**kwds) + return base_dict + + +class PrefixPathMapper(BasePathMapper): + match_type = 'prefix' + + def __init__(self, config): + super(PrefixPathMapper, self).__init__(config) + self.prefix_path = abspath(config['path']) + + def _path_matches(self, path): + return path.startswith(self.prefix_path) + + def to_pattern(self): + pattern_str = "(%s%s[^\s,\"\']+)" % (escape(self.prefix_path), escape(sep)) + return compile(pattern_str) + + def to_dict(self): + return self._extend_base_dict(path=self.prefix_path) + + +class GlobPathMapper(BasePathMapper): + match_type = 'glob' + + def __init__(self, config): + super(GlobPathMapper, self).__init__(config) + self.glob_path = config['path'] + + def _path_matches(self, path): + return fnmatch.fnmatch(path, self.glob_path) + + def to_pattern(self): + return compile(fnmatch.translate(self.glob_path)) + + def to_dict(self): + return self._extend_base_dict(path=self.glob_path) + + +class RegexPathMapper(BasePathMapper): + match_type = 'regex' + + def __init__(self, config): + super(RegexPathMapper, self).__init__(config) + self.pattern_raw = config['path'] + self.pattern = compile(self.pattern_raw) + + def _path_matches(self, path): + return self.pattern.match(path) is not None + + def to_pattern(self): + return self.pattern + + def to_dict(self): + return self._extend_base_dict(path=self.pattern_raw) + +MAPPER_CLASSES = [PrefixPathMapper, GlobPathMapper, RegexPathMapper] +MAPPER_CLASS_DICT = dict(map(lambda c: (c.match_type, c), MAPPER_CLASSES)) + + +def mappers_from_dicts(mapper_def_list): + return map(lambda m: __mappper_from_dict(m), mapper_def_list) + + +def __mappper_from_dict(mapper_dict): + map_type = mapper_dict.get('match_type', DEFAULT_PATH_MAPPER_TYPE) + return MAPPER_CLASS_DICT[map_type](mapper_dict) + + +class FileLister(object): + + def __init__(self, config): + self.depth = int(config.get("depth", "0")) + + def to_dict(self): + return dict( + depth=self.depth + ) + + def unstructured_map(self, path): + depth = self.depth + if self.depth == 0: + return {path: basename(path)} + else: + while depth > 0: + path = dirname(path) + depth -= 1 + return dict([(join(path, f), f) for f in directory_files(path)]) + +DEFAULT_FILE_LISTER = FileLister(dict(depth=0)) + +ACTION_CLASSES = [ + NoneAction, + RewriteAction, + TransferAction, + CopyAction, + RemoteCopyAction, + RemoteTransferAction, +] +actions = dict([(clazz.action_type, clazz) for clazz in ACTION_CLASSES]) + + +__all__ = [ + FileActionMapper, + path_type, + from_dict, + MessageAction, + RemoteTransferAction, # For testing +] diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/pulsar/client/amqp_exchange.py --- /dev/null +++ b/lib/pulsar/client/amqp_exchange.py @@ -0,0 +1,139 @@ +try: + import kombu + from kombu import pools +except ImportError: + kombu = None + +import socket +import logging +import threading +from time import sleep +log = logging.getLogger(__name__) + + +KOMBU_UNAVAILABLE = "Attempting to bind to AMQP message queue, but kombu dependency unavailable" + +DEFAULT_EXCHANGE_NAME = "pulsar" +DEFAULT_EXCHANGE_TYPE = "direct" +# Set timeout to periodically give up looking and check if polling should end. +DEFAULT_TIMEOUT = 0.2 +DEFAULT_HEARTBEAT = 580 + +DEFAULT_RECONNECT_CONSUMER_WAIT = 1 +DEFAULT_HEARTBEAT_WAIT = 1 + + +class PulsarExchange(object): + """ Utility for publishing and consuming structured Pulsar queues using kombu. + This is shared between the server and client - an exchange should be setup + for each manager (or in the case of the client, each manager one wished to + communicate with.) + + Each Pulsar manager is defined solely by name in the scheme, so only one Pulsar + should target each AMQP endpoint or care should be taken that unique + manager names are used across Pulsar servers targetting same AMQP endpoint - + and in particular only one such Pulsar should define an default manager with + name _default_. + """ + + def __init__( + self, + url, + manager_name, + connect_ssl=None, + timeout=DEFAULT_TIMEOUT, + publish_kwds={}, + ): + """ + """ + if not kombu: + raise Exception(KOMBU_UNAVAILABLE) + self.__url = url + self.__manager_name = manager_name + self.__connect_ssl = connect_ssl + self.__exchange = kombu.Exchange(DEFAULT_EXCHANGE_NAME, DEFAULT_EXCHANGE_TYPE) + self.__timeout = timeout + # Be sure to log message publishing failures. + if publish_kwds.get("retry", False): + if "retry_policy" not in publish_kwds: + publish_kwds["retry_policy"] = {} + if "errback" not in publish_kwds["retry_policy"]: + publish_kwds["retry_policy"]["errback"] = self.__publish_errback + self.__publish_kwds = publish_kwds + + @property + def url(self): + return self.__url + + def consume(self, queue_name, callback, check=True, connection_kwargs={}): + queue = self.__queue(queue_name) + log.debug("Consuming queue '%s'", queue) + while check: + heartbeat_thread = None + try: + with self.connection(self.__url, heartbeat=DEFAULT_HEARTBEAT, **connection_kwargs) as connection: + with kombu.Consumer(connection, queues=[queue], callbacks=[callback], accept=['json']): + heartbeat_thread = self.__start_heartbeat(queue_name, connection) + while check and connection.connected: + try: + connection.drain_events(timeout=self.__timeout) + except socket.timeout: + pass + except (IOError, socket.error), exc: + # In testing, errno is None + log.warning('Got %s, will retry: %s', exc.__class__.__name__, exc) + if heartbeat_thread: + heartbeat_thread.join() + sleep(DEFAULT_RECONNECT_CONSUMER_WAIT) + + def heartbeat(self, connection): + log.debug('AMQP heartbeat thread alive') + while connection.connected: + connection.heartbeat_check() + sleep(DEFAULT_HEARTBEAT_WAIT) + log.debug('AMQP heartbeat thread exiting') + + def publish(self, name, payload): + with self.connection(self.__url) as connection: + with pools.producers[connection].acquire() as producer: + key = self.__queue_name(name) + producer.publish( + payload, + serializer='json', + exchange=self.__exchange, + declare=[self.__exchange], + routing_key=key, + **self.__publish_kwds + ) + + def __publish_errback(self, exc, interval): + log.error("Connection error while publishing: %r", exc, exc_info=1) + log.info("Retrying in %s seconds", interval) + + def connection(self, connection_string, **kwargs): + if "ssl" not in kwargs: + kwargs["ssl"] = self.__connect_ssl + return kombu.Connection(connection_string, **kwargs) + + def __queue(self, name): + queue_name = self.__queue_name(name) + queue = kombu.Queue(queue_name, self.__exchange, routing_key=queue_name) + return queue + + def __queue_name(self, name): + key_prefix = self.__key_prefix() + queue_name = '%s_%s' % (key_prefix, name) + return queue_name + + def __key_prefix(self): + if self.__manager_name == "_default_": + key_prefix = "pulsar_" + else: + key_prefix = "pulsar_%s_" % self.__manager_name + return key_prefix + + def __start_heartbeat(self, queue_name, connection): + thread_name = "consume-heartbeat-%s" % (self.__queue_name(queue_name)) + thread = threading.Thread(name=thread_name, target=self.heartbeat, args=(connection,)) + thread.start() + return thread diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/pulsar/client/amqp_exchange_factory.py --- /dev/null +++ b/lib/pulsar/client/amqp_exchange_factory.py @@ -0,0 +1,41 @@ +from .amqp_exchange import PulsarExchange +from .util import filter_destination_params + + +def get_exchange(url, manager_name, params): + connect_ssl = parse_amqp_connect_ssl_params(params) + exchange_kwds = dict( + manager_name=manager_name, + connect_ssl=connect_ssl, + publish_kwds=parse_amqp_publish_kwds(params) + ) + timeout = params.get('amqp_consumer_timeout', False) + if timeout is not False: + exchange_kwds['timeout'] = timeout + exchange = PulsarExchange(url, **exchange_kwds) + return exchange + + +def parse_amqp_connect_ssl_params(params): + ssl_params = filter_destination_params(params, "amqp_connect_ssl_") + if not ssl_params: + return + + ssl = __import__('ssl') + if 'cert_reqs' in ssl_params: + value = ssl_params['cert_reqs'] + ssl_params['cert_reqs'] = getattr(ssl, value.upper()) + return ssl_params + + +def parse_amqp_publish_kwds(params): + all_publish_params = filter_destination_params(params, "amqp_publish_") + retry_policy_params = {} + for key in all_publish_params.keys(): + if key.startswith("retry_"): + value = all_publish_params[key] + retry_policy_params[key[len("retry_"):]] = value + del all_publish_params[key] + if retry_policy_params: + all_publish_params["retry_policy"] = retry_policy_params + return all_publish_params This diff is so big that we needed to truncate the remainder. Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 2 new changesets
by commits-noreply＠bitbucket.org 23 Jun '14

23 Jun '14

2 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/8c569cffa967/ Changeset: 8c569cffa967 User: jmchilton Date: 2014-06-20 07:28:36 Summary: Implement Pulsar job runners. LWR runner and client remains in Galaxy as is and can continue to target old LWR server. But these should be considered deprecated and will likely be dropped at some point. The LWR documentaiton in job_conf.xml.sample_advanced has been replaced with Pulsar documentation to reflect this. Pulsar runners will need to target a Pulsar server (now with a RESTful web interface - MQ option also available). Information on upgrading an LWR instance to pulsar can be found in the [Pulsar docs](http://pulsar.readthedocs.org/en/latest/#upgrading-from-the-lwr). Three new job runners have been added to replace the LWR runner. The `PulsarLegacyJobRunner` more or less has all of the old defaults that the LWR runner had and this one should be easiest to get working for people who are porting over old LWR servers (and will likely be most compatible with remote Windows hosts). The `PulsarRESTJobRunner` targets a remote Pulsar server over HTTP(S) but has newer standard configuration options such as rewriting parameters at tool stage by default, remote dependency evaluation by default, etc.... The `PulsarMQJobRunner` targets a remote Pulsar server via AMQP - and has the newer defaults of the RESTful runner as well as targetting the Galaxy job files API by default. The `url` runner parameter has been renamed `amqp_url` for clarity. These newer runners improve parameter handling in other ways such as warning deployers when ignored parameters are supplied. Affected #: 32 files diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 job_conf.xml.sample_advanced --- a/job_conf.xml.sample_advanced +++ b/job_conf.xml.sample_advanced @@ -19,27 +19,30 @@ <param id="drmaa_library_path">/sge/lib/libdrmaa.so</param></plugin> - <plugin id="lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner"> -  -  + <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" /> + <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" /> + <plugin id="slurm" type="runner" load="galaxy.jobs.runners.slurm:SlurmJobRunner" /> +  + <plugin id="pulsar_rest" type="runner" load="galaxy.jobs.runners.pulsar:PulsarRESTJobRunner"> +  -  - </plugin> - <plugin id="amqp_lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner"> - <param id="url">amqp://guest:guest@localhost:5672//</param> -  + <plugin id="pulsar_mq" type="runner" load="galaxy.jobs.runners.pulsar:PulsarMQJobRunner"> +  + <param id="amqp_url">amqp://guest:guest@localhost:5672//</param> + <param id="galaxy_url">http://localhost:8080</param> -  + </plugin> - <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" /> - <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" /> - <plugin id="slurm" type="runner" load="galaxy.jobs.runners.slurm:SlurmJobRunner" /> + <plugin id="pulsar_legacy" type="runner" load="galaxy.jobs.runners.pulsar:PulsarLegacyJobRunner"> +  + </plugin></plugins><handlers default="handlers"> @@ -213,87 +224,71 @@ <param id="function">foo</param></destination> - <destination id="secure_lwr" runner="lwr"> - <param id="url">https://windowshost.examle.com:8913/</param> -  + <destination id="secure_pulsar_rest_dest" runner="pulsar_rest"> + <param id="url">https://examle.com:8913/</param> + <param id="private_token">123456789changeme</param> -  -  +  -  -  -  +  -  -  -  +  +  + </destination> - <destination id="amqp_lwr_dest" runner="amqp_lwr" > -  -  - <param id="jobs_directory">/path/to/remote/lwr/lwr_staging/</param> -  - <param id="default_file_action">remote_transfer</param></destination><destination id="ssh_torque" runner="cli"><param id="shell_plugin">SecureShell</param> diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/galaxy/jobs/runners/pulsar.py --- /dev/null +++ b/lib/galaxy/jobs/runners/pulsar.py @@ -0,0 +1,707 @@ +from __future__ import absolute_import # Need to import pulsar_client absolutely. + +import logging + +from galaxy import model +from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner +from galaxy.jobs import ComputeEnvironment +from galaxy.jobs import JobDestination +from galaxy.jobs.command_factory import build_command +from galaxy.tools.deps import dependencies +from galaxy.util import string_as_bool_or_none +from galaxy.util.bunch import Bunch +from galaxy.util import specs + +import errno +from time import sleep +import os + +from pulsar.client import build_client_manager +from pulsar.client import url_to_destination_params +from pulsar.client import finish_job as pulsar_finish_job +from pulsar.client import submit_job as pulsar_submit_job +from pulsar.client import ClientJobDescription +from pulsar.client import PulsarOutputs +from pulsar.client import ClientOutputs +from pulsar.client import PathMapper + +log = logging.getLogger( __name__ ) + +__all__ = [ 'PulsarLegacyJobRunner', 'PulsarRESTJobRunner', 'PulsarMQJobRunner' ] + +NO_REMOTE_GALAXY_FOR_METADATA_MESSAGE = "Pulsar misconfiguration - Pulsar client configured to set metadata remotely, but remote Pulsar isn't properly configured with a galaxy_home directory." +NO_REMOTE_DATATYPES_CONFIG = "Pulsar client is configured to use remote datatypes configuration when setting metadata externally, but Pulsar is not configured with this information. Defaulting to datatypes_conf.xml." +GENERIC_REMOTE_ERROR = "Failed to communicate with remote job server." + +# Is there a good way to infer some default for this? Can only use +# url_for from web threads. https://gist.github.com/jmchilton/9098762 +DEFAULT_GALAXY_URL = "http://localhost:8080" + +PULSAR_PARAM_SPECS = dict( + transport=dict( + map=specs.to_str_or_none, + valid=specs.is_in("urllib", "curl", None), + default=None + ), + cache=dict( + map=specs.to_bool_or_none, + default=None, + ), + amqp_url=dict( + map=specs.to_str_or_none, + default=None, + ), + galaxy_url=dict( + map=specs.to_str_or_none, + default=DEFAULT_GALAXY_URL, + ), + manager=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_consumer_timeout=dict( + map=lambda val: None if val == "None" else float(val), + default=None, + ), + amqp_connect_ssl_ca_certs=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_keyfile=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_certfile=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_cert_reqs=dict( + map=specs.to_str_or_none, + default=None, + ), + # http://kombu.readthedocs.org/en/latest/reference/kombu.html#kombu.Producer.… + amqp_publish_retry=dict( + map=specs.to_bool, + default=False, + ), + amqp_publish_priority=dict( + map=int, + valid=lambda x: 0 <= x and x <= 9, + default=0, + ), + # http://kombu.readthedocs.org/en/latest/reference/kombu.html#kombu.Exchange.… + amqp_publish_delivery_mode=dict( + map=str, + valid=specs.is_in("transient", "persistent"), + default="persistent", + ), + amqp_publish_retry_max_retries=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_start=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_step=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_max=dict( + map=int, + default=None, + ), +) + + +PARAMETER_SPECIFICATION_REQUIRED = object() +PARAMETER_SPECIFICATION_IGNORED = object() + + +class PulsarJobRunner( AsynchronousJobRunner ): + """ + Pulsar Job Runner + """ + runner_name = "PulsarJobRunner" + + def __init__( self, app, nworkers, **kwds ): + """Start the job runner """ + super( PulsarJobRunner, self ).__init__( app, nworkers, runner_param_specs=PULSAR_PARAM_SPECS, **kwds ) + self._init_worker_threads() + galaxy_url = self.runner_params.galaxy_url + if galaxy_url: + galaxy_url = galaxy_url.rstrip("/") + self.galaxy_url = galaxy_url + self.__init_client_manager() + self._monitor() + + def _monitor( self ): + # Extension point allow MQ variant to setup callback instead + self._init_monitor_thread() + + def __init_client_manager( self ): + client_manager_kwargs = {} + for kwd in 'manager', 'cache', 'transport': + client_manager_kwargs[ kwd ] = self.runner_params[ kwd ] + for kwd in self.runner_params.keys(): + if kwd.startswith( 'amqp_' ): + client_manager_kwargs[ kwd ] = self.runner_params[ kwd ] + self.client_manager = build_client_manager(**client_manager_kwargs) + + def url_to_destination( self, url ): + """Convert a legacy URL to a job destination""" + return JobDestination( runner="pulsar", params=url_to_destination_params( url ) ) + + def check_watched_item(self, job_state): + try: + client = self.get_client_from_state(job_state) + status = client.get_status() + except Exception: + # An orphaned job was put into the queue at app startup, so remote server went down + # either way we are done I guess. + self.mark_as_finished(job_state) + return None + job_state = self._update_job_state_for_status(job_state, status) + return job_state + + def _update_job_state_for_status(self, job_state, pulsar_status): + if pulsar_status == "complete": + self.mark_as_finished(job_state) + return None + if pulsar_status == "failed": + self.fail_job(job_state) + return None + if pulsar_status == "running" and not job_state.running: + job_state.running = True + job_state.job_wrapper.change_state( model.Job.states.RUNNING ) + return job_state + + def queue_job(self, job_wrapper): + job_destination = job_wrapper.job_destination + self._populate_parameter_defaults( job_destination ) + + command_line, client, remote_job_config, compute_environment = self.__prepare_job( job_wrapper, job_destination ) + + if not command_line: + return + + try: + dependencies_description = PulsarJobRunner.__dependencies_description( client, job_wrapper ) + rewrite_paths = not PulsarJobRunner.__rewrite_parameters( client ) + unstructured_path_rewrites = {} + if compute_environment: + unstructured_path_rewrites = compute_environment.unstructured_path_rewrites + + client_job_description = ClientJobDescription( + command_line=command_line, + input_files=self.get_input_files(job_wrapper), + client_outputs=self.__client_outputs(client, job_wrapper), + working_directory=job_wrapper.working_directory, + tool=job_wrapper.tool, + config_files=job_wrapper.extra_filenames, + dependencies_description=dependencies_description, + env=client.env, + rewrite_paths=rewrite_paths, + arbitrary_files=unstructured_path_rewrites, + ) + job_id = pulsar_submit_job(client, client_job_description, remote_job_config) + log.info("Pulsar job submitted with job_id %s" % job_id) + job_wrapper.set_job_destination( job_destination, job_id ) + job_wrapper.change_state( model.Job.states.QUEUED ) + except Exception: + job_wrapper.fail( "failure running job", exception=True ) + log.exception("failure running job %d" % job_wrapper.job_id) + return + + pulsar_job_state = AsynchronousJobState() + pulsar_job_state.job_wrapper = job_wrapper + pulsar_job_state.job_id = job_id + pulsar_job_state.old_state = True + pulsar_job_state.running = False + pulsar_job_state.job_destination = job_destination + self.monitor_job(pulsar_job_state) + + def __prepare_job(self, job_wrapper, job_destination): + """ Build command-line and Pulsar client for this job. """ + command_line = None + client = None + remote_job_config = None + compute_environment = None + try: + client = self.get_client_from_wrapper(job_wrapper) + tool = job_wrapper.tool + remote_job_config = client.setup(tool.id, tool.version) + rewrite_parameters = PulsarJobRunner.__rewrite_parameters( client ) + prepare_kwds = {} + if rewrite_parameters: + compute_environment = PulsarComputeEnvironment( client, job_wrapper, remote_job_config ) + prepare_kwds[ 'compute_environment' ] = compute_environment + job_wrapper.prepare( **prepare_kwds ) + self.__prepare_input_files_locally(job_wrapper) + remote_metadata = PulsarJobRunner.__remote_metadata( client ) + dependency_resolution = PulsarJobRunner.__dependency_resolution( client ) + metadata_kwds = self.__build_metadata_configuration(client, job_wrapper, remote_metadata, remote_job_config) + remote_command_params = dict( + working_directory=remote_job_config['working_directory'], + metadata_kwds=metadata_kwds, + dependency_resolution=dependency_resolution, + ) + remote_working_directory = remote_job_config['working_directory'] + # TODO: Following defs work for Pulsar, always worked for Pulsar but should be + # calculated at some other level. + remote_job_directory = os.path.abspath(os.path.join(remote_working_directory, os.path.pardir)) + remote_tool_directory = os.path.abspath(os.path.join(remote_job_directory, "tool_files")) + container = self._find_container( + job_wrapper, + compute_working_directory=remote_working_directory, + compute_tool_directory=remote_tool_directory, + compute_job_directory=remote_job_directory, + ) + command_line = build_command( + self, + job_wrapper=job_wrapper, + container=container, + include_metadata=remote_metadata, + include_work_dir_outputs=False, + remote_command_params=remote_command_params, + ) + except Exception: + job_wrapper.fail( "failure preparing job", exception=True ) + log.exception("failure running job %d" % job_wrapper.job_id) + + # If we were able to get a command line, run the job + if not command_line: + job_wrapper.finish( '', '' ) + + return command_line, client, remote_job_config, compute_environment + + def __prepare_input_files_locally(self, job_wrapper): + """Run task splitting commands locally.""" + prepare_input_files_cmds = getattr(job_wrapper, 'prepare_input_files_cmds', None) + if prepare_input_files_cmds is not None: + for cmd in prepare_input_files_cmds: # run the commands to stage the input files + if 0 != os.system(cmd): + raise Exception('Error running file staging command: %s' % cmd) + job_wrapper.prepare_input_files_cmds = None # prevent them from being used in-line + + def _populate_parameter_defaults( self, job_destination ): + updated = False + params = job_destination.params + for key, value in self.destination_defaults.iteritems(): + if key in params: + if value is PARAMETER_SPECIFICATION_IGNORED: + log.warn( "Pulsar runner in selected configuration ignores parameter %s" % key ) + continue + #if self.runner_params.get( key, None ): + # # Let plugin define defaults for some parameters - + # # for instance that way jobs_directory can be + # # configured next to AMQP url (where it belongs). + # params[ key ] = self.runner_params[ key ] + # continue + + if not value: + continue + + if value is PARAMETER_SPECIFICATION_REQUIRED: + raise Exception( "Pulsar destination does not define required parameter %s" % key ) + elif value is not PARAMETER_SPECIFICATION_IGNORED: + params[ key ] = value + updated = True + return updated + + def get_output_files(self, job_wrapper): + output_paths = job_wrapper.get_output_fnames() + return [ str( o ) for o in output_paths ] # Force job_path from DatasetPath objects. + + def get_input_files(self, job_wrapper): + input_paths = job_wrapper.get_input_paths() + return [ str( i ) for i in input_paths ] # Force job_path from DatasetPath objects. + + def get_client_from_wrapper(self, job_wrapper): + job_id = job_wrapper.job_id + if hasattr(job_wrapper, 'task_id'): + job_id = "%s_%s" % (job_id, job_wrapper.task_id) + params = job_wrapper.job_destination.params.copy() + for key, value in params.iteritems(): + if value: + params[key] = model.User.expand_user_properties( job_wrapper.get_job().user, value ) + + 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, env=[] ): + # Cannot use url_for outside of web thread. + #files_endpoint = url_for( controller="job_files", job_id=encoded_job_id ) + + encoded_job_id = self.app.security.encode_id(job_id) + job_key = self.app.security.encode_id( job_id, kind="jobs_files" ) + files_endpoint = "%s/api/jobs/%s/files?job_key=%s" % ( + self.galaxy_url, + encoded_job_id, + job_key + ) + 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 ) + + def finish_job( self, job_state ): + stderr = stdout = '' + job_wrapper = job_state.job_wrapper + try: + client = self.get_client_from_state(job_state) + run_results = client.full_status() + remote_working_directory = run_results.get("working_directory", None) + stdout = run_results.get('stdout', '') + stderr = run_results.get('stderr', '') + exit_code = run_results.get('returncode', None) + pulsar_outputs = PulsarOutputs.from_status_response(run_results) + # Use Pulsar client code to transfer/copy files back + # and cleanup job if needed. + completed_normally = \ + job_wrapper.get_state() not in [ model.Job.states.ERROR, model.Job.states.DELETED ] + cleanup_job = self.app.config.cleanup_job + client_outputs = self.__client_outputs(client, job_wrapper) + finish_args = dict( client=client, + job_completed_normally=completed_normally, + cleanup_job=cleanup_job, + client_outputs=client_outputs, + pulsar_outputs=pulsar_outputs ) + failed = pulsar_finish_job( **finish_args ) + if failed: + job_wrapper.fail("Failed to find or download one or more job outputs from remote server.", exception=True) + except Exception: + message = GENERIC_REMOTE_ERROR + job_wrapper.fail( message, exception=True ) + log.exception("failure finishing job %d" % job_wrapper.job_id) + return + if not PulsarJobRunner.__remote_metadata( client ): + self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) + # Finish the job + try: + job_wrapper.finish( + stdout, + stderr, + exit_code, + remote_working_directory=remote_working_directory + ) + except Exception: + log.exception("Job wrapper finish method failed") + job_wrapper.fail("Unable to finish job", exception=True) + + def fail_job( self, job_state ): + """ + Seperated out so we can use the worker threads for it. + """ + self.stop_job( self.sa_session.query( self.app.model.Job ).get( job_state.job_wrapper.job_id ) ) + job_state.job_wrapper.fail( getattr( job_state, "fail_message", GENERIC_REMOTE_ERROR ) ) + + def check_pid( self, pid ): + try: + os.kill( pid, 0 ) + return True + except OSError, e: + if e.errno == errno.ESRCH: + log.debug( "check_pid(): PID %d is dead" % pid ) + else: + log.warning( "check_pid(): Got errno %s when attempting to check PID %d: %s" % ( errno.errorcode[e.errno], pid, e.strerror ) ) + return False + + def stop_job( self, job ): + #if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished + job_ext_output_metadata = job.get_external_output_metadata() + if job_ext_output_metadata: + pid = job_ext_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them + if pid in [ None, '' ]: + log.warning( "stop_job(): %s: no PID in database for job, unable to stop" % job.id ) + return + pid = int( pid ) + if not self.check_pid( pid ): + log.warning( "stop_job(): %s: PID %d was already dead or can't be signaled" % ( job.id, pid ) ) + return + for sig in [ 15, 9 ]: + try: + os.killpg( pid, sig ) + except OSError, e: + log.warning( "stop_job(): %s: Got errno %s when attempting to signal %d to PID %d: %s" % ( job.id, errno.errorcode[e.errno], sig, pid, e.strerror ) ) + return # give up + sleep( 2 ) + if not self.check_pid( pid ): + log.debug( "stop_job(): %s: PID %d successfully killed with signal %d" % ( job.id, pid, sig ) ) + return + else: + log.warning( "stop_job(): %s: PID %d refuses to die after signaling TERM/KILL" % ( job.id, pid ) ) + else: + # Remote kill + pulsar_url = job.job_runner_name + job_id = job.job_runner_external_id + log.debug("Attempt remote Pulsar kill of job with url %s and id %s" % (pulsar_url, job_id)) + client = self.get_client(job.destination_params, job_id) + client.kill() + + def recover( self, job, job_wrapper ): + """Recovers jobs stuck in the queued/running state when Galaxy started""" + job_state = self._job_state( job, job_wrapper ) + job_wrapper.command_line = job.get_command_line() + state = job.get_state() + if state in [model.Job.states.RUNNING, model.Job.states.QUEUED]: + log.debug( "(Pulsar/%s) is still in running state, adding to the Pulsar queue" % ( job.get_id()) ) + job_state.old_state = True + job_state.running = state == model.Job.states.RUNNING + self.monitor_queue.put( job_state ) + + def shutdown( self ): + super( PulsarJobRunner, self ).shutdown() + self.client_manager.shutdown() + + def _job_state( self, job, job_wrapper ): + job_state = AsynchronousJobState() + # TODO: Determine why this is set when using normal message queue updates + # but not CLI submitted MQ updates... + raw_job_id = job.get_job_runner_external_id() or job_wrapper.job_id + job_state.job_id = str( raw_job_id ) + job_state.runner_url = job_wrapper.get_job_runner_url() + job_state.job_destination = job_wrapper.job_destination + job_state.job_wrapper = job_wrapper + return job_state + + def __client_outputs( self, client, job_wrapper ): + work_dir_outputs = self.get_work_dir_outputs( job_wrapper ) + output_files = self.get_output_files( job_wrapper ) + client_outputs = ClientOutputs( + working_directory=job_wrapper.working_directory, + work_dir_outputs=work_dir_outputs, + output_files=output_files, + version_file=job_wrapper.get_version_string_path(), + ) + return client_outputs + + @staticmethod + def __dependencies_description( pulsar_client, job_wrapper ): + dependency_resolution = PulsarJobRunner.__dependency_resolution( pulsar_client ) + remote_dependency_resolution = dependency_resolution == "remote" + if not remote_dependency_resolution: + return None + requirements = job_wrapper.tool.requirements or [] + installed_tool_dependencies = job_wrapper.tool.installed_tool_dependencies or [] + return dependencies.DependenciesDescription( + requirements=requirements, + installed_tool_dependencies=installed_tool_dependencies, + ) + + @staticmethod + def __dependency_resolution( pulsar_client ): + dependency_resolution = pulsar_client.destination_params.get( "dependency_resolution", "local" ) + if dependency_resolution not in ["none", "local", "remote"]: + raise Exception("Unknown dependency_resolution value encountered %s" % dependency_resolution) + return dependency_resolution + + @staticmethod + def __remote_metadata( pulsar_client ): + remote_metadata = string_as_bool_or_none( pulsar_client.destination_params.get( "remote_metadata", False ) ) + return remote_metadata + + @staticmethod + def __use_remote_datatypes_conf( pulsar_client ): + """ When setting remote metadata, use integrated datatypes from this + Galaxy instance or use the datatypes config configured via the remote + Pulsar. + + Both options are broken in different ways for same reason - datatypes + may not match. One can push the local datatypes config to the remote + server - but there is no guarentee these datatypes will be defined + there. Alternatively, one can use the remote datatype config - but + there is no guarentee that it will contain all the datatypes available + to this Galaxy. + """ + use_remote_datatypes = string_as_bool_or_none( pulsar_client.destination_params.get( "use_remote_datatypes", False ) ) + return use_remote_datatypes + + @staticmethod + def __rewrite_parameters( pulsar_client ): + return string_as_bool_or_none( pulsar_client.destination_params.get( "rewrite_parameters", False ) ) or False + + def __build_metadata_configuration(self, client, job_wrapper, remote_metadata, remote_job_config): + metadata_kwds = {} + if remote_metadata: + remote_system_properties = remote_job_config.get("system_properties", {}) + remote_galaxy_home = remote_system_properties.get("galaxy_home", None) + if not remote_galaxy_home: + raise Exception(NO_REMOTE_GALAXY_FOR_METADATA_MESSAGE) + metadata_kwds['exec_dir'] = remote_galaxy_home + outputs_directory = remote_job_config['outputs_directory'] + configs_directory = remote_job_config['configs_directory'] + working_directory = remote_job_config['working_directory'] + # For metadata calculation, we need to build a list of of output + # file objects with real path indicating location on Galaxy server + # and false path indicating location on compute server. Since the + # Pulsar disables from_work_dir copying as part of the job command + # line we need to take the list of output locations on the Pulsar + # server (produced by self.get_output_files(job_wrapper)) and for + # each work_dir output substitute the effective path on the Pulsar + # server relative to the remote working directory as the + # false_path to send the metadata command generation module. + work_dir_outputs = self.get_work_dir_outputs(job_wrapper, job_working_directory=working_directory) + outputs = [Bunch(false_path=os.path.join(outputs_directory, os.path.basename(path)), real_path=path) for path in self.get_output_files(job_wrapper)] + for output in outputs: + for pulsar_workdir_path, real_path in work_dir_outputs: + if real_path == output.real_path: + output.false_path = pulsar_workdir_path + metadata_kwds['output_fnames'] = outputs + metadata_kwds['compute_tmp_dir'] = working_directory + metadata_kwds['config_root'] = remote_galaxy_home + default_config_file = os.path.join(remote_galaxy_home, 'universe_wsgi.ini') + metadata_kwds['config_file'] = remote_system_properties.get('galaxy_config_file', default_config_file) + metadata_kwds['dataset_files_path'] = remote_system_properties.get('galaxy_dataset_files_path', None) + if PulsarJobRunner.__use_remote_datatypes_conf( client ): + remote_datatypes_config = remote_system_properties.get('galaxy_datatypes_config_file', None) + if not remote_datatypes_config: + log.warn(NO_REMOTE_DATATYPES_CONFIG) + remote_datatypes_config = os.path.join(remote_galaxy_home, 'datatypes_conf.xml') + metadata_kwds['datatypes_config'] = remote_datatypes_config + else: + integrates_datatypes_config = self.app.datatypes_registry.integrated_datatypes_configs + # Ensure this file gets pushed out to the remote config dir. + job_wrapper.extra_filenames.append(integrates_datatypes_config) + + metadata_kwds['datatypes_config'] = os.path.join(configs_directory, os.path.basename(integrates_datatypes_config)) + return metadata_kwds + + +class PulsarLegacyJobRunner( PulsarJobRunner ): + destination_defaults = dict( + rewrite_parameters="false", + dependency_resolution="local", + ) + + +class PulsarMQJobRunner( PulsarJobRunner ): + destination_defaults = dict( + default_file_action="remote_transfer", + rewrite_parameters="true", + dependency_resolution="remote", + jobs_directory=PARAMETER_SPECIFICATION_REQUIRED, + url=PARAMETER_SPECIFICATION_IGNORED, + private_token=PARAMETER_SPECIFICATION_IGNORED + ) + + def _monitor( self ): + # This is a message queue driven runner, don't monitor + # just setup required callback. + self.client_manager.ensure_has_status_update_callback(self.__async_update) + + def __async_update( self, full_status ): + job_id = None + try: + job_id = full_status[ "job_id" ] + job, job_wrapper = self.app.job_manager.job_handler.job_queue.job_pair_for_id( job_id ) + job_state = self._job_state( job, job_wrapper ) + self._update_job_state_for_status(job_state, full_status[ "status" ] ) + except Exception: + log.exception( "Failed to update Pulsar job status for job_id %s" % job_id ) + raise + # Nothing else to do? - Attempt to fail the job? + + +class PulsarRESTJobRunner( PulsarJobRunner ): + destination_defaults = dict( + default_file_action="transfer", + rewrite_parameters="true", + dependency_resolution="remote", + url=PARAMETER_SPECIFICATION_REQUIRED, + ) + + +class PulsarComputeEnvironment( ComputeEnvironment ): + + def __init__( self, pulsar_client, job_wrapper, remote_job_config ): + self.pulsar_client = pulsar_client + self.job_wrapper = job_wrapper + self.local_path_config = job_wrapper.default_compute_environment() + self.unstructured_path_rewrites = {} + # job_wrapper.prepare is going to expunge the job backing the following + # computations, so precalculate these paths. + self._wrapper_input_paths = self.local_path_config.input_paths() + self._wrapper_output_paths = self.local_path_config.output_paths() + self.path_mapper = PathMapper(pulsar_client, remote_job_config, self.local_path_config.working_directory()) + self._config_directory = remote_job_config[ "configs_directory" ] + self._working_directory = remote_job_config[ "working_directory" ] + self._sep = remote_job_config[ "system_properties" ][ "separator" ] + self._tool_dir = remote_job_config[ "tools_directory" ] + version_path = self.local_path_config.version_path() + new_version_path = self.path_mapper.remote_version_path_rewrite(version_path) + if new_version_path: + version_path = new_version_path + self._version_path = version_path + + def output_paths( self ): + local_output_paths = self._wrapper_output_paths + + results = [] + for local_output_path in local_output_paths: + wrapper_path = str( local_output_path ) + remote_path = self.path_mapper.remote_output_path_rewrite( wrapper_path ) + results.append( self._dataset_path( local_output_path, remote_path ) ) + return results + + def input_paths( self ): + local_input_paths = self._wrapper_input_paths + + results = [] + for local_input_path in local_input_paths: + wrapper_path = str( local_input_path ) + # This will over-copy in some cases. For instance in the case of task + # splitting, this input will be copied even though only the work dir + # input will actually be used. + remote_path = self.path_mapper.remote_input_path_rewrite( wrapper_path ) + results.append( self._dataset_path( local_input_path, remote_path ) ) + return results + + def _dataset_path( self, local_dataset_path, remote_path ): + remote_extra_files_path = None + if remote_path: + remote_extra_files_path = "%s_files" % remote_path[ 0:-len( ".dat" ) ] + return local_dataset_path.with_path_for_job( remote_path, remote_extra_files_path ) + + def working_directory( self ): + return self._working_directory + + def config_directory( self ): + return self._config_directory + + def new_file_path( self ): + return self.working_directory() # Problems with doing this? + + def sep( self ): + return self._sep + + def version_path( self ): + return self._version_path + + def rewriter( self, parameter_value ): + unstructured_path_rewrites = self.unstructured_path_rewrites + if parameter_value in unstructured_path_rewrites: + # Path previously mapped, use previous mapping. + return unstructured_path_rewrites[ parameter_value ] + if parameter_value in unstructured_path_rewrites.itervalues(): + # Path is a rewritten remote path (this might never occur, + # consider dropping check...) + return parameter_value + + rewrite, new_unstructured_path_rewrites = self.path_mapper.check_for_arbitrary_rewrite( parameter_value ) + if rewrite: + unstructured_path_rewrites.update(new_unstructured_path_rewrites) + return rewrite + else: + # Did need to rewrite, use original path or value. + return parameter_value + + def unstructured_path_rewriter( self ): + return self.rewriter diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/galaxy/jobs/runners/util/__init__.py --- a/lib/galaxy/jobs/runners/util/__init__.py +++ b/lib/galaxy/jobs/runners/util/__init__.py @@ -1,7 +1,7 @@ """ This module and its submodules contains utilities for running external processes and interfacing with job managers. This module should contain -functionality shared between Galaxy and the LWR. +functionality shared between Galaxy and the Pulsar. """ from galaxy.util.bunch import Bunch diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/galaxy/jobs/runners/util/cli/factory.py --- a/lib/galaxy/jobs/runners/util/cli/factory.py +++ b/lib/galaxy/jobs/runners/util/cli/factory.py @@ -5,7 +5,7 @@ ) code_dir = 'lib' except ImportError: - from lwr.managers.util.cli import ( + from pulsar.managers.util.cli import ( CliInterface, split_params ) diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/galaxy/jobs/runners/util/cli/job/slurm.py --- a/lib/galaxy/jobs/runners/util/cli/job/slurm.py +++ b/lib/galaxy/jobs/runners/util/cli/job/slurm.py @@ -5,7 +5,7 @@ from galaxy.model import Job job_states = Job.states except ImportError: - # Not in Galaxy, map Galaxy job states to LWR ones. + # Not in Galaxy, map Galaxy job states to Pulsar ones. from galaxy.util import enum job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/galaxy/jobs/runners/util/cli/job/torque.py --- a/lib/galaxy/jobs/runners/util/cli/job/torque.py +++ b/lib/galaxy/jobs/runners/util/cli/job/torque.py @@ -7,7 +7,7 @@ from galaxy.model import Job job_states = Job.states except ImportError: - # Not in Galaxy, map Galaxy job states to LWR ones. + # Not in Galaxy, map Galaxy job states to Pulsar ones. from galaxy.util import enum job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/galaxy/objectstore/__init__.py --- a/lib/galaxy/objectstore/__init__.py +++ b/lib/galaxy/objectstore/__init__.py @@ -623,9 +623,9 @@ elif store == 'irods': from .rods import IRODSObjectStore return IRODSObjectStore(config=config, config_xml=config_xml) - elif store == 'lwr': - from .lwr import LwrObjectStore - return LwrObjectStore(config=config, config_xml=config_xml) + elif store == 'pulsar': + from .pulsar import PulsarObjectStore + return PulsarObjectStore(config=config, config_xml=config_xml) else: log.error("Unrecognized object store definition: {0}".format(store)) diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/galaxy/objectstore/lwr.py --- a/lib/galaxy/objectstore/lwr.py +++ /dev/null @@ -1,76 +0,0 @@ -from __future__ import absolute_import # Need to import lwr_client absolutely. -from ..objectstore import ObjectStore -try: - from galaxy.jobs.runners.lwr_client.manager import ObjectStoreClientManager -except ImportError: - from lwr.lwr_client.manager import ObjectStoreClientManager - - -class LwrObjectStore(ObjectStore): - """ - Object store implementation that delegates to a remote LWR server. - - This may be more aspirational than practical for now, it would be good to - Galaxy to a point that a handler thread could be setup that doesn't attempt - to access the disk files returned by a (this) object store - just passing - them along to the LWR unmodified. That modification - along with this - implementation and LWR job destinations would then allow Galaxy to fully - manage jobs on remote servers with completely different mount points. - - This implementation should be considered beta and may be dropped from - Galaxy at some future point or significantly modified. - """ - - def __init__(self, config, config_xml): - self.lwr_client = self.__build_lwr_client(config_xml) - - def exists(self, obj, **kwds): - return self.lwr_client.exists(**self.__build_kwds(obj, **kwds)) - - def file_ready(self, obj, **kwds): - return self.lwr_client.file_ready(**self.__build_kwds(obj, **kwds)) - - def create(self, obj, **kwds): - return self.lwr_client.create(**self.__build_kwds(obj, **kwds)) - - def empty(self, obj, **kwds): - return self.lwr_client.empty(**self.__build_kwds(obj, **kwds)) - - def size(self, obj, **kwds): - return self.lwr_client.size(**self.__build_kwds(obj, **kwds)) - - def delete(self, obj, **kwds): - return self.lwr_client.delete(**self.__build_kwds(obj, **kwds)) - - # TODO: Optimize get_data. - def get_data(self, obj, **kwds): - return self.lwr_client.get_data(**self.__build_kwds(obj, **kwds)) - - def get_filename(self, obj, **kwds): - return self.lwr_client.get_filename(**self.__build_kwds(obj, **kwds)) - - def update_from_file(self, obj, **kwds): - return self.lwr_client.update_from_file(**self.__build_kwds(obj, **kwds)) - - def get_store_usage_percent(self): - return self.lwr_client.get_store_usage_percent() - - def get_object_url(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None): - return None - - def __build_kwds(self, obj, **kwds): - kwds['object_id'] = obj.id - return kwds - pass - - def __build_lwr_client(self, config_xml): - url = config_xml.get("url") - private_token = config_xml.get("private_token", None) - transport = config_xml.get("transport", None) - manager_options = dict(transport=transport) - client_options = dict(url=url, private_token=private_token) - lwr_client = ObjectStoreClientManager(**manager_options).get_client(client_options) - return lwr_client - - def shutdown(self): - pass diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/galaxy/objectstore/pulsar.py --- /dev/null +++ b/lib/galaxy/objectstore/pulsar.py @@ -0,0 +1,73 @@ +from __future__ import absolute_import # Need to import pulsar_client absolutely. +from ..objectstore import ObjectStore +from pulsar.client.manager import ObjectStoreClientManager + + +class PulsarObjectStore(ObjectStore): + """ + Object store implementation that delegates to a remote Pulsar server. + + This may be more aspirational than practical for now, it would be good to + Galaxy to a point that a handler thread could be setup that doesn't attempt + to access the disk files returned by a (this) object store - just passing + them along to the Pulsar unmodified. That modification - along with this + implementation and Pulsar job destinations would then allow Galaxy to fully + manage jobs on remote servers with completely different mount points. + + This implementation should be considered beta and may be dropped from + Galaxy at some future point or significantly modified. + """ + + def __init__(self, config, config_xml): + self.pulsar_client = self.__build_pulsar_client(config_xml) + + def exists(self, obj, **kwds): + return self.pulsar_client.exists(**self.__build_kwds(obj, **kwds)) + + def file_ready(self, obj, **kwds): + return self.pulsar_client.file_ready(**self.__build_kwds(obj, **kwds)) + + def create(self, obj, **kwds): + return self.pulsar_client.create(**self.__build_kwds(obj, **kwds)) + + def empty(self, obj, **kwds): + return self.pulsar_client.empty(**self.__build_kwds(obj, **kwds)) + + def size(self, obj, **kwds): + return self.pulsar_client.size(**self.__build_kwds(obj, **kwds)) + + def delete(self, obj, **kwds): + return self.pulsar_client.delete(**self.__build_kwds(obj, **kwds)) + + # TODO: Optimize get_data. + def get_data(self, obj, **kwds): + return self.pulsar_client.get_data(**self.__build_kwds(obj, **kwds)) + + def get_filename(self, obj, **kwds): + return self.pulsar_client.get_filename(**self.__build_kwds(obj, **kwds)) + + def update_from_file(self, obj, **kwds): + return self.pulsar_client.update_from_file(**self.__build_kwds(obj, **kwds)) + + def get_store_usage_percent(self): + return self.pulsar_client.get_store_usage_percent() + + def get_object_url(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None): + return None + + def __build_kwds(self, obj, **kwds): + kwds['object_id'] = obj.id + return kwds + pass + + def __build_pulsar_client(self, config_xml): + url = config_xml.get("url") + private_token = config_xml.get("private_token", None) + transport = config_xml.get("transport", None) + manager_options = dict(transport=transport) + client_options = dict(url=url, private_token=private_token) + pulsar_client = ObjectStoreClientManager(**manager_options).get_client(client_options) + return pulsar_client + + def shutdown(self): + pass diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/galaxy/tools/deps/dependencies.py --- a/lib/galaxy/tools/deps/dependencies.py +++ b/lib/galaxy/tools/deps/dependencies.py @@ -8,7 +8,7 @@ related context required to resolve dependencies via the ToolShedPackageDependencyResolver. - This is meant to enable remote resolution of dependencies, by the LWR or + This is meant to enable remote resolution of dependencies, by the Pulsar or other potential remote execution mechanisms. """ @@ -39,7 +39,7 @@ @staticmethod def _toolshed_install_dependency_from_dict(as_dict): - # Rather than requiring full models in LWR, just use simple objects + # Rather than requiring full models in Pulsar, just use simple objects # containing only properties and associations used to resolve # dependencies for tool execution. repository_object = bunch.Bunch( diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/pulsar/client/__init__.py --- /dev/null +++ b/lib/pulsar/client/__init__.py @@ -0,0 +1,62 @@ +""" +pulsar client +====== + +This module contains logic for interfacing with an external Pulsar server. + +------------------ +Configuring Galaxy +------------------ + +Galaxy job runners are configured in Galaxy's ``job_conf.xml`` file. See ``job_conf.xml.sample_advanced`` +in your Galaxy code base or on +`Bitbucket <https://bitbucket.org/galaxy/galaxy-dist/src/tip/job_conf.xml.sample_advanc…>`_ +for information on how to configure Galaxy to interact with the Pulsar. + +Galaxy also supports an older, less rich configuration of job runners directly +in its main ``universe_wsgi.ini`` file. The following section describes how to +configure Galaxy to communicate with the Pulsar in this legacy mode. + +Legacy +------ + +A Galaxy tool can be configured to be executed remotely via Pulsar by +adding a line to the ``universe_wsgi.ini`` file under the +``galaxy:tool_runners`` section with the format:: + + <tool_id> = pulsar://http://<pulsar_host>:<pulsar_port> + +As an example, if a host named remotehost is running the Pulsar server +application on port ``8913``, then the tool with id ``test_tool`` can +be configured to run remotely on remotehost by adding the following +line to ``universe.ini``:: + + test_tool = pulsar://http://remotehost:8913 + +Remember this must be added after the ``[galaxy:tool_runners]`` header +in the ``universe.ini`` file. + + +""" + +from .staging.down import finish_job +from .staging.up import submit_job +from .staging import ClientJobDescription +from .staging import PulsarOutputs +from .staging import ClientOutputs +from .client import OutputNotFoundException +from .manager import build_client_manager +from .destination import url_to_destination_params +from .path_mapper import PathMapper + +__all__ = [ + build_client_manager, + OutputNotFoundException, + url_to_destination_params, + finish_job, + submit_job, + ClientJobDescription, + PulsarOutputs, + ClientOutputs, + PathMapper, +] diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/pulsar/client/action_mapper.py --- /dev/null +++ b/lib/pulsar/client/action_mapper.py @@ -0,0 +1,567 @@ +from json import load +from os import makedirs +from os.path import exists +from os.path import abspath +from os.path import dirname +from os.path import join +from os.path import basename +from os.path import sep +import fnmatch +from re import compile +from re import escape +import galaxy.util +from galaxy.util.bunch import Bunch +from .config_util import read_file +from .util import directory_files +from .util import unique_path_prefix +from .transport import get_file +from .transport import post_file + + +DEFAULT_MAPPED_ACTION = 'transfer' # Not really clear to me what this should be, exception? +DEFAULT_PATH_MAPPER_TYPE = 'prefix' + +STAGING_ACTION_REMOTE = "remote" +STAGING_ACTION_LOCAL = "local" +STAGING_ACTION_NONE = None +STAGING_ACTION_DEFAULT = "default" + +# Poor man's enum. +path_type = Bunch( + # Galaxy input datasets and extra files. + INPUT="input", + # Galaxy config and param files. + CONFIG="config", + # Files from tool's tool_dir (for now just wrapper if available). + TOOL="tool", + # Input work dir files - e.g. metadata files, task-split input files, etc.. + WORKDIR="workdir", + # Galaxy output datasets in their final home. + OUTPUT="output", + # Galaxy from_work_dir output paths and other files (e.g. galaxy.json) + OUTPUT_WORKDIR="output_workdir", + # Other fixed tool parameter paths (likely coming from tool data, but not + # nessecarily). Not sure this is the best name... + UNSTRUCTURED="unstructured", +) + + +ACTION_DEFAULT_PATH_TYPES = [ + path_type.INPUT, + path_type.CONFIG, + path_type.TOOL, + path_type.WORKDIR, + path_type.OUTPUT, + path_type.OUTPUT_WORKDIR, +] +ALL_PATH_TYPES = ACTION_DEFAULT_PATH_TYPES + [path_type.UNSTRUCTURED] + + +class FileActionMapper(object): + """ + Objects of this class define how paths are mapped to actions. + + >>> json_string = r'''{"paths": [ \ + {"path": "/opt/galaxy", "action": "none"}, \ + {"path": "/galaxy/data", "action": "transfer"}, \ + {"path": "/cool/bamfiles/**/*.bam", "action": "copy", "match_type": "glob"}, \ + {"path": ".*/dataset_\\\\d+.dat", "action": "copy", "match_type": "regex"} \ + ]}''' + >>> from tempfile import NamedTemporaryFile + >>> from os import unlink + >>> def mapper_for(default_action, config_contents): + ... f = NamedTemporaryFile(delete=False) + ... f.write(config_contents.encode('UTF-8')) + ... f.close() + ... mock_client = Bunch(default_file_action=default_action, action_config_path=f.name, files_endpoint=None) + ... mapper = FileActionMapper(mock_client) + ... mapper = FileActionMapper(config=mapper.to_dict()) # Serialize and deserialize it to make sure still works + ... unlink(f.name) + ... return mapper + >>> mapper = mapper_for(default_action='none', config_contents=json_string) + >>> # Test first config line above, implicit path prefix mapper + >>> action = mapper.action('/opt/galaxy/tools/filters/catWrapper.py', 'input') + >>> action.action_type == u'none' + True + >>> action.staging_needed + False + >>> # Test another (2nd) mapper, this one with a different action + >>> action = mapper.action('/galaxy/data/files/000/dataset_1.dat', 'input') + >>> action.action_type == u'transfer' + True + >>> action.staging_needed + True + >>> # Always at least copy work_dir outputs. + >>> action = mapper.action('/opt/galaxy/database/working_directory/45.sh', 'workdir') + >>> action.action_type == u'copy' + True + >>> action.staging_needed + True + >>> # Test glob mapper (matching test) + >>> mapper.action('/cool/bamfiles/projectABC/study1/patient3.bam', 'input').action_type == u'copy' + True + >>> # Test glob mapper (non-matching test) + >>> mapper.action('/cool/bamfiles/projectABC/study1/patient3.bam.bai', 'input').action_type == u'none' + True + >>> # Regex mapper test. + >>> mapper.action('/old/galaxy/data/dataset_10245.dat', 'input').action_type == u'copy' + True + >>> # Doesn't map unstructured paths by default + >>> mapper.action('/old/galaxy/data/dataset_10245.dat', 'unstructured').action_type == u'none' + True + >>> input_only_mapper = mapper_for(default_action="none", config_contents=r'''{"paths": [ \ + {"path": "/", "action": "transfer", "path_types": "input"} \ + ] }''') + >>> input_only_mapper.action('/dataset_1.dat', 'input').action_type == u'transfer' + True + >>> input_only_mapper.action('/dataset_1.dat', 'output').action_type == u'none' + True + >>> unstructured_mapper = mapper_for(default_action="none", config_contents=r'''{"paths": [ \ + {"path": "/", "action": "transfer", "path_types": "*any*"} \ + ] }''') + >>> unstructured_mapper.action('/old/galaxy/data/dataset_10245.dat', 'unstructured').action_type == u'transfer' + True + """ + + def __init__(self, client=None, config=None): + if config is None and client is None: + message = "FileActionMapper must be constructed from either a client or a config dictionary." + raise Exception(message) + if config is None: + config = self.__client_to_config(client) + self.default_action = config.get("default_action", "transfer") + self.mappers = mappers_from_dicts(config.get("paths", [])) + self.files_endpoint = config.get("files_endpoint", None) + + def action(self, path, type, mapper=None): + mapper = self.__find_mapper(path, type, mapper) + action_class = self.__action_class(path, type, mapper) + file_lister = DEFAULT_FILE_LISTER + action_kwds = {} + if mapper: + file_lister = mapper.file_lister + action_kwds = mapper.action_kwds + action = action_class(path, file_lister=file_lister, **action_kwds) + self.__process_action(action, type) + return action + + def unstructured_mappers(self): + """ Return mappers that will map 'unstructured' files (i.e. go beyond + mapping inputs, outputs, and config files). + """ + return filter(lambda m: path_type.UNSTRUCTURED in m.path_types, self.mappers) + + def to_dict(self): + return dict( + default_action=self.default_action, + files_endpoint=self.files_endpoint, + paths=map(lambda m: m.to_dict(), self.mappers) + ) + + def __client_to_config(self, client): + action_config_path = client.action_config_path + if action_config_path: + config = read_file(action_config_path) + else: + config = dict() + config["default_action"] = client.default_file_action + config["files_endpoint"] = client.files_endpoint + return config + + def __load_action_config(self, path): + config = load(open(path, 'rb')) + self.mappers = mappers_from_dicts(config.get('paths', [])) + + def __find_mapper(self, path, type, mapper=None): + if not mapper: + normalized_path = abspath(path) + for query_mapper in self.mappers: + if query_mapper.matches(normalized_path, type): + mapper = query_mapper + break + return mapper + + def __action_class(self, path, type, mapper): + action_type = self.default_action if type in ACTION_DEFAULT_PATH_TYPES else "none" + if mapper: + action_type = mapper.action_type + if type in ["workdir", "output_workdir"] and action_type == "none": + # We are changing the working_directory relative to what + # Galaxy would use, these need to be copied over. + action_type = "copy" + action_class = actions.get(action_type, None) + if action_class is None: + message_template = "Unknown action_type encountered %s while trying to map path %s" + message_args = (action_type, path) + raise Exception(message_template % message_args) + return action_class + + def __process_action(self, action, file_type): + """ Extension point to populate extra action information after an + action has been created. + """ + if action.action_type == "remote_transfer": + url_base = self.files_endpoint + if not url_base: + raise Exception("Attempted to use remote_transfer action with defining a files_endpoint") + if "?" not in url_base: + url_base = "%s?" % url_base + # TODO: URL encode path. + url = "%s&path=%s&file_type=%s" % (url_base, action.path, file_type) + action.url = url + +REQUIRED_ACTION_KWD = object() + + +class BaseAction(object): + action_spec = {} + + def __init__(self, path, file_lister=None): + self.path = path + self.file_lister = file_lister or DEFAULT_FILE_LISTER + + def unstructured_map(self, path_helper): + unstructured_map = self.file_lister.unstructured_map(self.path) + if self.staging_needed: + # To ensure uniqueness, prepend unique prefix to each name + prefix = unique_path_prefix(self.path) + for path, name in unstructured_map.iteritems(): + unstructured_map[path] = join(prefix, name) + else: + path_rewrites = {} + for path in unstructured_map: + rewrite = self.path_rewrite(path_helper, path) + if rewrite: + path_rewrites[path] = rewrite + unstructured_map = path_rewrites + return unstructured_map + + @property + def staging_needed(self): + return self.staging != STAGING_ACTION_NONE + + @property + def staging_action_local(self): + return self.staging == STAGING_ACTION_LOCAL + + +class NoneAction(BaseAction): + """ This action indicates the corresponding path does not require any + additional action. This should indicate paths that are available both on + the Pulsar client (i.e. Galaxy server) and remote Pulsar server with the same + paths. """ + action_type = "none" + staging = STAGING_ACTION_NONE + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type) + + @classmethod + def from_dict(cls, action_dict): + return NoneAction(path=action_dict["path"]) + + def path_rewrite(self, path_helper, path=None): + return None + + +class RewriteAction(BaseAction): + """ This actin indicates the Pulsar server should simply rewrite the path + to the specified file. + """ + action_spec = dict( + source_directory=REQUIRED_ACTION_KWD, + destination_directory=REQUIRED_ACTION_KWD + ) + action_type = "rewrite" + staging = STAGING_ACTION_NONE + + def __init__(self, path, file_lister=None, source_directory=None, destination_directory=None): + self.path = path + self.file_lister = file_lister or DEFAULT_FILE_LISTER + self.source_directory = source_directory + self.destination_directory = destination_directory + + def to_dict(self): + return dict( + path=self.path, + action_type=self.action_type, + source_directory=self.source_directory, + destination_directory=self.destination_directory, + ) + + @classmethod + def from_dict(cls, action_dict): + return RewriteAction( + path=action_dict["path"], + source_directory=action_dict["source_directory"], + destination_directory=action_dict["destination_directory"], + ) + + def path_rewrite(self, path_helper, path=None): + if not path: + path = self.path + new_path = path_helper.from_posix_with_new_base(self.path, self.source_directory, self.destination_directory) + return None if new_path == self.path else new_path + + +class TransferAction(BaseAction): + """ This actions indicates that the Pulsar client should initiate an HTTP + transfer of the corresponding path to the remote Pulsar server before + launching the job. """ + action_type = "transfer" + staging = STAGING_ACTION_LOCAL + + +class CopyAction(BaseAction): + """ This action indicates that the Pulsar client should execute a file system + copy of the corresponding path to the Pulsar staging directory prior to + launching the corresponding job. """ + action_type = "copy" + staging = STAGING_ACTION_LOCAL + + +class RemoteCopyAction(BaseAction): + """ This action indicates the Pulsar server should copy the file before + execution via direct file system copy. This is like a CopyAction, but + it indicates the action should occur on the Pulsar server instead of on + the client. + """ + action_type = "remote_copy" + staging = STAGING_ACTION_REMOTE + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type) + + @classmethod + def from_dict(cls, action_dict): + return RemoteCopyAction(path=action_dict["path"]) + + def write_to_path(self, path): + galaxy.util.copy_to_path(open(self.path, "rb"), path) + + def write_from_path(self, pulsar_path): + destination = self.path + parent_directory = dirname(destination) + if not exists(parent_directory): + makedirs(parent_directory) + with open(pulsar_path, "rb") as f: + galaxy.util.copy_to_path(f, destination) + + +class RemoteTransferAction(BaseAction): + """ This action indicates the Pulsar server should copy the file before + execution via direct file system copy. This is like a CopyAction, but + it indicates the action should occur on the Pulsar server instead of on + the client. + """ + action_type = "remote_transfer" + staging = STAGING_ACTION_REMOTE + + def __init__(self, path, file_lister=None, url=None): + super(RemoteTransferAction, self).__init__(path, file_lister=file_lister) + self.url = url + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type, url=self.url) + + @classmethod + def from_dict(cls, action_dict): + return RemoteTransferAction(path=action_dict["path"], url=action_dict["url"]) + + def write_to_path(self, path): + get_file(self.url, path) + + def write_from_path(self, pulsar_path): + post_file(self.url, pulsar_path) + + +class MessageAction(object): + """ Sort of pseudo action describing "files" store in memory and + transferred via message (HTTP, Python-call, MQ, etc...) + """ + action_type = "message" + staging = STAGING_ACTION_DEFAULT + + def __init__(self, contents, client=None): + self.contents = contents + self.client = client + + @property + def staging_needed(self): + return True + + @property + def staging_action_local(self): + # Ekkk, cannot be called if created through from_dict. + # Shouldn't be a problem the way it is used - but is an + # object design problem. + return self.client.prefer_local_staging + + def to_dict(self): + return dict(contents=self.contents, action_type=MessageAction.action_type) + + @classmethod + def from_dict(cls, action_dict): + return MessageAction(contents=action_dict["contents"]) + + def write_to_path(self, path): + open(path, "w").write(self.contents) + + +DICTIFIABLE_ACTION_CLASSES = [RemoteCopyAction, RemoteTransferAction, MessageAction] + + +def from_dict(action_dict): + action_type = action_dict.get("action_type", None) + target_class = None + for action_class in DICTIFIABLE_ACTION_CLASSES: + if action_type == action_class.action_type: + target_class = action_class + if not target_class: + message = "Failed to recover action from dictionary - invalid action type specified %s." % action_type + raise Exception(message) + return target_class.from_dict(action_dict) + + +class BasePathMapper(object): + + def __init__(self, config): + action_type = config.get('action', DEFAULT_MAPPED_ACTION) + action_class = actions.get(action_type, None) + action_kwds = action_class.action_spec.copy() + for key, value in action_kwds.items(): + if key in config: + action_kwds[key] = config[key] + elif value is REQUIRED_ACTION_KWD: + message_template = "action_type %s requires key word argument %s" + message = message_template % (action_type, key) + raise Exception(message) + self.action_type = action_type + self.action_kwds = action_kwds + path_types_str = config.get('path_types', "*defaults*") + path_types_str = path_types_str.replace("*defaults*", ",".join(ACTION_DEFAULT_PATH_TYPES)) + path_types_str = path_types_str.replace("*any*", ",".join(ALL_PATH_TYPES)) + self.path_types = path_types_str.split(",") + self.file_lister = FileLister(config) + + def matches(self, path, path_type): + path_type_matches = path_type in self.path_types + return path_type_matches and self._path_matches(path) + + def _extend_base_dict(self, **kwds): + base_dict = dict( + action=self.action_type, + path_types=",".join(self.path_types), + match_type=self.match_type + ) + base_dict.update(self.file_lister.to_dict()) + base_dict.update(self.action_kwds) + base_dict.update(**kwds) + return base_dict + + +class PrefixPathMapper(BasePathMapper): + match_type = 'prefix' + + def __init__(self, config): + super(PrefixPathMapper, self).__init__(config) + self.prefix_path = abspath(config['path']) + + def _path_matches(self, path): + return path.startswith(self.prefix_path) + + def to_pattern(self): + pattern_str = "(%s%s[^\s,\"\']+)" % (escape(self.prefix_path), escape(sep)) + return compile(pattern_str) + + def to_dict(self): + return self._extend_base_dict(path=self.prefix_path) + + +class GlobPathMapper(BasePathMapper): + match_type = 'glob' + + def __init__(self, config): + super(GlobPathMapper, self).__init__(config) + self.glob_path = config['path'] + + def _path_matches(self, path): + return fnmatch.fnmatch(path, self.glob_path) + + def to_pattern(self): + return compile(fnmatch.translate(self.glob_path)) + + def to_dict(self): + return self._extend_base_dict(path=self.glob_path) + + +class RegexPathMapper(BasePathMapper): + match_type = 'regex' + + def __init__(self, config): + super(RegexPathMapper, self).__init__(config) + self.pattern_raw = config['path'] + self.pattern = compile(self.pattern_raw) + + def _path_matches(self, path): + return self.pattern.match(path) is not None + + def to_pattern(self): + return self.pattern + + def to_dict(self): + return self._extend_base_dict(path=self.pattern_raw) + +MAPPER_CLASSES = [PrefixPathMapper, GlobPathMapper, RegexPathMapper] +MAPPER_CLASS_DICT = dict(map(lambda c: (c.match_type, c), MAPPER_CLASSES)) + + +def mappers_from_dicts(mapper_def_list): + return map(lambda m: __mappper_from_dict(m), mapper_def_list) + + +def __mappper_from_dict(mapper_dict): + map_type = mapper_dict.get('match_type', DEFAULT_PATH_MAPPER_TYPE) + return MAPPER_CLASS_DICT[map_type](mapper_dict) + + +class FileLister(object): + + def __init__(self, config): + self.depth = int(config.get("depth", "0")) + + def to_dict(self): + return dict( + depth=self.depth + ) + + def unstructured_map(self, path): + depth = self.depth + if self.depth == 0: + return {path: basename(path)} + else: + while depth > 0: + path = dirname(path) + depth -= 1 + return dict([(join(path, f), f) for f in directory_files(path)]) + +DEFAULT_FILE_LISTER = FileLister(dict(depth=0)) + +ACTION_CLASSES = [ + NoneAction, + RewriteAction, + TransferAction, + CopyAction, + RemoteCopyAction, + RemoteTransferAction, +] +actions = dict([(clazz.action_type, clazz) for clazz in ACTION_CLASSES]) + + +__all__ = [ + FileActionMapper, + path_type, + from_dict, + MessageAction, + RemoteTransferAction, # For testing +] diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/pulsar/client/amqp_exchange.py --- /dev/null +++ b/lib/pulsar/client/amqp_exchange.py @@ -0,0 +1,139 @@ +try: + import kombu + from kombu import pools +except ImportError: + kombu = None + +import socket +import logging +import threading +from time import sleep +log = logging.getLogger(__name__) + + +KOMBU_UNAVAILABLE = "Attempting to bind to AMQP message queue, but kombu dependency unavailable" + +DEFAULT_EXCHANGE_NAME = "pulsar" +DEFAULT_EXCHANGE_TYPE = "direct" +# Set timeout to periodically give up looking and check if polling should end. +DEFAULT_TIMEOUT = 0.2 +DEFAULT_HEARTBEAT = 580 + +DEFAULT_RECONNECT_CONSUMER_WAIT = 1 +DEFAULT_HEARTBEAT_WAIT = 1 + + +class PulsarExchange(object): + """ Utility for publishing and consuming structured Pulsar queues using kombu. + This is shared between the server and client - an exchange should be setup + for each manager (or in the case of the client, each manager one wished to + communicate with.) + + Each Pulsar manager is defined solely by name in the scheme, so only one Pulsar + should target each AMQP endpoint or care should be taken that unique + manager names are used across Pulsar servers targetting same AMQP endpoint - + and in particular only one such Pulsar should define an default manager with + name _default_. + """ + + def __init__( + self, + url, + manager_name, + connect_ssl=None, + timeout=DEFAULT_TIMEOUT, + publish_kwds={}, + ): + """ + """ + if not kombu: + raise Exception(KOMBU_UNAVAILABLE) + self.__url = url + self.__manager_name = manager_name + self.__connect_ssl = connect_ssl + self.__exchange = kombu.Exchange(DEFAULT_EXCHANGE_NAME, DEFAULT_EXCHANGE_TYPE) + self.__timeout = timeout + # Be sure to log message publishing failures. + if publish_kwds.get("retry", False): + if "retry_policy" not in publish_kwds: + publish_kwds["retry_policy"] = {} + if "errback" not in publish_kwds["retry_policy"]: + publish_kwds["retry_policy"]["errback"] = self.__publish_errback + self.__publish_kwds = publish_kwds + + @property + def url(self): + return self.__url + + def consume(self, queue_name, callback, check=True, connection_kwargs={}): + queue = self.__queue(queue_name) + log.debug("Consuming queue '%s'", queue) + while check: + heartbeat_thread = None + try: + with self.connection(self.__url, heartbeat=DEFAULT_HEARTBEAT, **connection_kwargs) as connection: + with kombu.Consumer(connection, queues=[queue], callbacks=[callback], accept=['json']): + heartbeat_thread = self.__start_heartbeat(queue_name, connection) + while check and connection.connected: + try: + connection.drain_events(timeout=self.__timeout) + except socket.timeout: + pass + except (IOError, socket.error), exc: + # In testing, errno is None + log.warning('Got %s, will retry: %s', exc.__class__.__name__, exc) + if heartbeat_thread: + heartbeat_thread.join() + sleep(DEFAULT_RECONNECT_CONSUMER_WAIT) + + def heartbeat(self, connection): + log.debug('AMQP heartbeat thread alive') + while connection.connected: + connection.heartbeat_check() + sleep(DEFAULT_HEARTBEAT_WAIT) + log.debug('AMQP heartbeat thread exiting') + + def publish(self, name, payload): + with self.connection(self.__url) as connection: + with pools.producers[connection].acquire() as producer: + key = self.__queue_name(name) + producer.publish( + payload, + serializer='json', + exchange=self.__exchange, + declare=[self.__exchange], + routing_key=key, + **self.__publish_kwds + ) + + def __publish_errback(self, exc, interval): + log.error("Connection error while publishing: %r", exc, exc_info=1) + log.info("Retrying in %s seconds", interval) + + def connection(self, connection_string, **kwargs): + if "ssl" not in kwargs: + kwargs["ssl"] = self.__connect_ssl + return kombu.Connection(connection_string, **kwargs) + + def __queue(self, name): + queue_name = self.__queue_name(name) + queue = kombu.Queue(queue_name, self.__exchange, routing_key=queue_name) + return queue + + def __queue_name(self, name): + key_prefix = self.__key_prefix() + queue_name = '%s_%s' % (key_prefix, name) + return queue_name + + def __key_prefix(self): + if self.__manager_name == "_default_": + key_prefix = "pulsar_" + else: + key_prefix = "pulsar_%s_" % self.__manager_name + return key_prefix + + def __start_heartbeat(self, queue_name, connection): + thread_name = "consume-heartbeat-%s" % (self.__queue_name(queue_name)) + thread = threading.Thread(name=thread_name, target=self.heartbeat, args=(connection,)) + thread.start() + return thread diff -r cfa24d46767a41afa9e83cbf703eef837b47f1e2 -r 8c569cffa9670f765492c8ddc11fb54d9a2ab2f1 lib/pulsar/client/amqp_exchange_factory.py --- /dev/null +++ b/lib/pulsar/client/amqp_exchange_factory.py @@ -0,0 +1,41 @@ +from .amqp_exchange import PulsarExchange +from .util import filter_destination_params + + +def get_exchange(url, manager_name, params): + connect_ssl = parse_amqp_connect_ssl_params(params) + exchange_kwds = dict( + manager_name=manager_name, + connect_ssl=connect_ssl, + publish_kwds=parse_amqp_publish_kwds(params) + ) + timeout = params.get('amqp_consumer_timeout', False) + if timeout is not False: + exchange_kwds['timeout'] = timeout + exchange = PulsarExchange(url, **exchange_kwds) + return exchange + + +def parse_amqp_connect_ssl_params(params): + ssl_params = filter_destination_params(params, "amqp_connect_ssl_") + if not ssl_params: + return + + ssl = __import__('ssl') + if 'cert_reqs' in ssl_params: + value = ssl_params['cert_reqs'] + ssl_params['cert_reqs'] = getattr(ssl, value.upper()) + return ssl_params + + +def parse_amqp_publish_kwds(params): + all_publish_params = filter_destination_params(params, "amqp_publish_") + retry_policy_params = {} + for key in all_publish_params.keys(): + if key.startswith("retry_"): + value = all_publish_params[key] + retry_policy_params[key[len("retry_"):]] = value + del all_publish_params[key] + if retry_policy_params: + all_publish_params["retry_policy"] = retry_policy_params + return all_publish_params This diff is so big that we needed to truncate the remainder. https://bitbucket.org/galaxy/galaxy-central/commits/566388d62374/ Changeset: 566388d62374 User: jmchilton Date: 2014-06-24 04:27:51 Summary: Merged in jmchilton/galaxy-central-fork-1 (pull request #422) Implement Pulsar job runners. Affected #: 32 files diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 job_conf.xml.sample_advanced --- a/job_conf.xml.sample_advanced +++ b/job_conf.xml.sample_advanced @@ -19,27 +19,30 @@ <param id="drmaa_library_path">/sge/lib/libdrmaa.so</param></plugin> - <plugin id="lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner"> -  -  + <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" /> + <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" /> + <plugin id="slurm" type="runner" load="galaxy.jobs.runners.slurm:SlurmJobRunner" /> +  + <plugin id="pulsar_rest" type="runner" load="galaxy.jobs.runners.pulsar:PulsarRESTJobRunner"> +  -  - </plugin> - <plugin id="amqp_lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner"> - <param id="url">amqp://guest:guest@localhost:5672//</param> -  + <plugin id="pulsar_mq" type="runner" load="galaxy.jobs.runners.pulsar:PulsarMQJobRunner"> +  + <param id="amqp_url">amqp://guest:guest@localhost:5672//</param> + <param id="galaxy_url">http://localhost:8080</param> -  + </plugin> - <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" /> - <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" /> - <plugin id="slurm" type="runner" load="galaxy.jobs.runners.slurm:SlurmJobRunner" /> + <plugin id="pulsar_legacy" type="runner" load="galaxy.jobs.runners.pulsar:PulsarLegacyJobRunner"> +  + </plugin></plugins><handlers default="handlers"> @@ -213,87 +224,71 @@ <param id="function">foo</param></destination> - <destination id="secure_lwr" runner="lwr"> - <param id="url">https://windowshost.examle.com:8913/</param> -  + <destination id="secure_pulsar_rest_dest" runner="pulsar_rest"> + <param id="url">https://examle.com:8913/</param> + <param id="private_token">123456789changeme</param> -  -  +  -  -  -  +  -  -  -  +  +  + </destination> - <destination id="amqp_lwr_dest" runner="amqp_lwr" > -  -  - <param id="jobs_directory">/path/to/remote/lwr/lwr_staging/</param> -  - <param id="default_file_action">remote_transfer</param></destination><destination id="ssh_torque" runner="cli"><param id="shell_plugin">SecureShell</param> diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/pulsar.py --- /dev/null +++ b/lib/galaxy/jobs/runners/pulsar.py @@ -0,0 +1,707 @@ +from __future__ import absolute_import # Need to import pulsar_client absolutely. + +import logging + +from galaxy import model +from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner +from galaxy.jobs import ComputeEnvironment +from galaxy.jobs import JobDestination +from galaxy.jobs.command_factory import build_command +from galaxy.tools.deps import dependencies +from galaxy.util import string_as_bool_or_none +from galaxy.util.bunch import Bunch +from galaxy.util import specs + +import errno +from time import sleep +import os + +from pulsar.client import build_client_manager +from pulsar.client import url_to_destination_params +from pulsar.client import finish_job as pulsar_finish_job +from pulsar.client import submit_job as pulsar_submit_job +from pulsar.client import ClientJobDescription +from pulsar.client import PulsarOutputs +from pulsar.client import ClientOutputs +from pulsar.client import PathMapper + +log = logging.getLogger( __name__ ) + +__all__ = [ 'PulsarLegacyJobRunner', 'PulsarRESTJobRunner', 'PulsarMQJobRunner' ] + +NO_REMOTE_GALAXY_FOR_METADATA_MESSAGE = "Pulsar misconfiguration - Pulsar client configured to set metadata remotely, but remote Pulsar isn't properly configured with a galaxy_home directory." +NO_REMOTE_DATATYPES_CONFIG = "Pulsar client is configured to use remote datatypes configuration when setting metadata externally, but Pulsar is not configured with this information. Defaulting to datatypes_conf.xml." +GENERIC_REMOTE_ERROR = "Failed to communicate with remote job server." + +# Is there a good way to infer some default for this? Can only use +# url_for from web threads. https://gist.github.com/jmchilton/9098762 +DEFAULT_GALAXY_URL = "http://localhost:8080" + +PULSAR_PARAM_SPECS = dict( + transport=dict( + map=specs.to_str_or_none, + valid=specs.is_in("urllib", "curl", None), + default=None + ), + cache=dict( + map=specs.to_bool_or_none, + default=None, + ), + amqp_url=dict( + map=specs.to_str_or_none, + default=None, + ), + galaxy_url=dict( + map=specs.to_str_or_none, + default=DEFAULT_GALAXY_URL, + ), + manager=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_consumer_timeout=dict( + map=lambda val: None if val == "None" else float(val), + default=None, + ), + amqp_connect_ssl_ca_certs=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_keyfile=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_certfile=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_cert_reqs=dict( + map=specs.to_str_or_none, + default=None, + ), + # http://kombu.readthedocs.org/en/latest/reference/kombu.html#kombu.Producer.… + amqp_publish_retry=dict( + map=specs.to_bool, + default=False, + ), + amqp_publish_priority=dict( + map=int, + valid=lambda x: 0 <= x and x <= 9, + default=0, + ), + # http://kombu.readthedocs.org/en/latest/reference/kombu.html#kombu.Exchange.… + amqp_publish_delivery_mode=dict( + map=str, + valid=specs.is_in("transient", "persistent"), + default="persistent", + ), + amqp_publish_retry_max_retries=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_start=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_step=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_max=dict( + map=int, + default=None, + ), +) + + +PARAMETER_SPECIFICATION_REQUIRED = object() +PARAMETER_SPECIFICATION_IGNORED = object() + + +class PulsarJobRunner( AsynchronousJobRunner ): + """ + Pulsar Job Runner + """ + runner_name = "PulsarJobRunner" + + def __init__( self, app, nworkers, **kwds ): + """Start the job runner """ + super( PulsarJobRunner, self ).__init__( app, nworkers, runner_param_specs=PULSAR_PARAM_SPECS, **kwds ) + self._init_worker_threads() + galaxy_url = self.runner_params.galaxy_url + if galaxy_url: + galaxy_url = galaxy_url.rstrip("/") + self.galaxy_url = galaxy_url + self.__init_client_manager() + self._monitor() + + def _monitor( self ): + # Extension point allow MQ variant to setup callback instead + self._init_monitor_thread() + + def __init_client_manager( self ): + client_manager_kwargs = {} + for kwd in 'manager', 'cache', 'transport': + client_manager_kwargs[ kwd ] = self.runner_params[ kwd ] + for kwd in self.runner_params.keys(): + if kwd.startswith( 'amqp_' ): + client_manager_kwargs[ kwd ] = self.runner_params[ kwd ] + self.client_manager = build_client_manager(**client_manager_kwargs) + + def url_to_destination( self, url ): + """Convert a legacy URL to a job destination""" + return JobDestination( runner="pulsar", params=url_to_destination_params( url ) ) + + def check_watched_item(self, job_state): + try: + client = self.get_client_from_state(job_state) + status = client.get_status() + except Exception: + # An orphaned job was put into the queue at app startup, so remote server went down + # either way we are done I guess. + self.mark_as_finished(job_state) + return None + job_state = self._update_job_state_for_status(job_state, status) + return job_state + + def _update_job_state_for_status(self, job_state, pulsar_status): + if pulsar_status == "complete": + self.mark_as_finished(job_state) + return None + if pulsar_status == "failed": + self.fail_job(job_state) + return None + if pulsar_status == "running" and not job_state.running: + job_state.running = True + job_state.job_wrapper.change_state( model.Job.states.RUNNING ) + return job_state + + def queue_job(self, job_wrapper): + job_destination = job_wrapper.job_destination + self._populate_parameter_defaults( job_destination ) + + command_line, client, remote_job_config, compute_environment = self.__prepare_job( job_wrapper, job_destination ) + + if not command_line: + return + + try: + dependencies_description = PulsarJobRunner.__dependencies_description( client, job_wrapper ) + rewrite_paths = not PulsarJobRunner.__rewrite_parameters( client ) + unstructured_path_rewrites = {} + if compute_environment: + unstructured_path_rewrites = compute_environment.unstructured_path_rewrites + + client_job_description = ClientJobDescription( + command_line=command_line, + input_files=self.get_input_files(job_wrapper), + client_outputs=self.__client_outputs(client, job_wrapper), + working_directory=job_wrapper.working_directory, + tool=job_wrapper.tool, + config_files=job_wrapper.extra_filenames, + dependencies_description=dependencies_description, + env=client.env, + rewrite_paths=rewrite_paths, + arbitrary_files=unstructured_path_rewrites, + ) + job_id = pulsar_submit_job(client, client_job_description, remote_job_config) + log.info("Pulsar job submitted with job_id %s" % job_id) + job_wrapper.set_job_destination( job_destination, job_id ) + job_wrapper.change_state( model.Job.states.QUEUED ) + except Exception: + job_wrapper.fail( "failure running job", exception=True ) + log.exception("failure running job %d" % job_wrapper.job_id) + return + + pulsar_job_state = AsynchronousJobState() + pulsar_job_state.job_wrapper = job_wrapper + pulsar_job_state.job_id = job_id + pulsar_job_state.old_state = True + pulsar_job_state.running = False + pulsar_job_state.job_destination = job_destination + self.monitor_job(pulsar_job_state) + + def __prepare_job(self, job_wrapper, job_destination): + """ Build command-line and Pulsar client for this job. """ + command_line = None + client = None + remote_job_config = None + compute_environment = None + try: + client = self.get_client_from_wrapper(job_wrapper) + tool = job_wrapper.tool + remote_job_config = client.setup(tool.id, tool.version) + rewrite_parameters = PulsarJobRunner.__rewrite_parameters( client ) + prepare_kwds = {} + if rewrite_parameters: + compute_environment = PulsarComputeEnvironment( client, job_wrapper, remote_job_config ) + prepare_kwds[ 'compute_environment' ] = compute_environment + job_wrapper.prepare( **prepare_kwds ) + self.__prepare_input_files_locally(job_wrapper) + remote_metadata = PulsarJobRunner.__remote_metadata( client ) + dependency_resolution = PulsarJobRunner.__dependency_resolution( client ) + metadata_kwds = self.__build_metadata_configuration(client, job_wrapper, remote_metadata, remote_job_config) + remote_command_params = dict( + working_directory=remote_job_config['working_directory'], + metadata_kwds=metadata_kwds, + dependency_resolution=dependency_resolution, + ) + remote_working_directory = remote_job_config['working_directory'] + # TODO: Following defs work for Pulsar, always worked for Pulsar but should be + # calculated at some other level. + remote_job_directory = os.path.abspath(os.path.join(remote_working_directory, os.path.pardir)) + remote_tool_directory = os.path.abspath(os.path.join(remote_job_directory, "tool_files")) + container = self._find_container( + job_wrapper, + compute_working_directory=remote_working_directory, + compute_tool_directory=remote_tool_directory, + compute_job_directory=remote_job_directory, + ) + command_line = build_command( + self, + job_wrapper=job_wrapper, + container=container, + include_metadata=remote_metadata, + include_work_dir_outputs=False, + remote_command_params=remote_command_params, + ) + except Exception: + job_wrapper.fail( "failure preparing job", exception=True ) + log.exception("failure running job %d" % job_wrapper.job_id) + + # If we were able to get a command line, run the job + if not command_line: + job_wrapper.finish( '', '' ) + + return command_line, client, remote_job_config, compute_environment + + def __prepare_input_files_locally(self, job_wrapper): + """Run task splitting commands locally.""" + prepare_input_files_cmds = getattr(job_wrapper, 'prepare_input_files_cmds', None) + if prepare_input_files_cmds is not None: + for cmd in prepare_input_files_cmds: # run the commands to stage the input files + if 0 != os.system(cmd): + raise Exception('Error running file staging command: %s' % cmd) + job_wrapper.prepare_input_files_cmds = None # prevent them from being used in-line + + def _populate_parameter_defaults( self, job_destination ): + updated = False + params = job_destination.params + for key, value in self.destination_defaults.iteritems(): + if key in params: + if value is PARAMETER_SPECIFICATION_IGNORED: + log.warn( "Pulsar runner in selected configuration ignores parameter %s" % key ) + continue + #if self.runner_params.get( key, None ): + # # Let plugin define defaults for some parameters - + # # for instance that way jobs_directory can be + # # configured next to AMQP url (where it belongs). + # params[ key ] = self.runner_params[ key ] + # continue + + if not value: + continue + + if value is PARAMETER_SPECIFICATION_REQUIRED: + raise Exception( "Pulsar destination does not define required parameter %s" % key ) + elif value is not PARAMETER_SPECIFICATION_IGNORED: + params[ key ] = value + updated = True + return updated + + def get_output_files(self, job_wrapper): + output_paths = job_wrapper.get_output_fnames() + return [ str( o ) for o in output_paths ] # Force job_path from DatasetPath objects. + + def get_input_files(self, job_wrapper): + input_paths = job_wrapper.get_input_paths() + return [ str( i ) for i in input_paths ] # Force job_path from DatasetPath objects. + + def get_client_from_wrapper(self, job_wrapper): + job_id = job_wrapper.job_id + if hasattr(job_wrapper, 'task_id'): + job_id = "%s_%s" % (job_id, job_wrapper.task_id) + params = job_wrapper.job_destination.params.copy() + for key, value in params.iteritems(): + if value: + params[key] = model.User.expand_user_properties( job_wrapper.get_job().user, value ) + + 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, env=[] ): + # Cannot use url_for outside of web thread. + #files_endpoint = url_for( controller="job_files", job_id=encoded_job_id ) + + encoded_job_id = self.app.security.encode_id(job_id) + job_key = self.app.security.encode_id( job_id, kind="jobs_files" ) + files_endpoint = "%s/api/jobs/%s/files?job_key=%s" % ( + self.galaxy_url, + encoded_job_id, + job_key + ) + 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 ) + + def finish_job( self, job_state ): + stderr = stdout = '' + job_wrapper = job_state.job_wrapper + try: + client = self.get_client_from_state(job_state) + run_results = client.full_status() + remote_working_directory = run_results.get("working_directory", None) + stdout = run_results.get('stdout', '') + stderr = run_results.get('stderr', '') + exit_code = run_results.get('returncode', None) + pulsar_outputs = PulsarOutputs.from_status_response(run_results) + # Use Pulsar client code to transfer/copy files back + # and cleanup job if needed. + completed_normally = \ + job_wrapper.get_state() not in [ model.Job.states.ERROR, model.Job.states.DELETED ] + cleanup_job = self.app.config.cleanup_job + client_outputs = self.__client_outputs(client, job_wrapper) + finish_args = dict( client=client, + job_completed_normally=completed_normally, + cleanup_job=cleanup_job, + client_outputs=client_outputs, + pulsar_outputs=pulsar_outputs ) + failed = pulsar_finish_job( **finish_args ) + if failed: + job_wrapper.fail("Failed to find or download one or more job outputs from remote server.", exception=True) + except Exception: + message = GENERIC_REMOTE_ERROR + job_wrapper.fail( message, exception=True ) + log.exception("failure finishing job %d" % job_wrapper.job_id) + return + if not PulsarJobRunner.__remote_metadata( client ): + self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) + # Finish the job + try: + job_wrapper.finish( + stdout, + stderr, + exit_code, + remote_working_directory=remote_working_directory + ) + except Exception: + log.exception("Job wrapper finish method failed") + job_wrapper.fail("Unable to finish job", exception=True) + + def fail_job( self, job_state ): + """ + Seperated out so we can use the worker threads for it. + """ + self.stop_job( self.sa_session.query( self.app.model.Job ).get( job_state.job_wrapper.job_id ) ) + job_state.job_wrapper.fail( getattr( job_state, "fail_message", GENERIC_REMOTE_ERROR ) ) + + def check_pid( self, pid ): + try: + os.kill( pid, 0 ) + return True + except OSError, e: + if e.errno == errno.ESRCH: + log.debug( "check_pid(): PID %d is dead" % pid ) + else: + log.warning( "check_pid(): Got errno %s when attempting to check PID %d: %s" % ( errno.errorcode[e.errno], pid, e.strerror ) ) + return False + + def stop_job( self, job ): + #if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished + job_ext_output_metadata = job.get_external_output_metadata() + if job_ext_output_metadata: + pid = job_ext_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them + if pid in [ None, '' ]: + log.warning( "stop_job(): %s: no PID in database for job, unable to stop" % job.id ) + return + pid = int( pid ) + if not self.check_pid( pid ): + log.warning( "stop_job(): %s: PID %d was already dead or can't be signaled" % ( job.id, pid ) ) + return + for sig in [ 15, 9 ]: + try: + os.killpg( pid, sig ) + except OSError, e: + log.warning( "stop_job(): %s: Got errno %s when attempting to signal %d to PID %d: %s" % ( job.id, errno.errorcode[e.errno], sig, pid, e.strerror ) ) + return # give up + sleep( 2 ) + if not self.check_pid( pid ): + log.debug( "stop_job(): %s: PID %d successfully killed with signal %d" % ( job.id, pid, sig ) ) + return + else: + log.warning( "stop_job(): %s: PID %d refuses to die after signaling TERM/KILL" % ( job.id, pid ) ) + else: + # Remote kill + pulsar_url = job.job_runner_name + job_id = job.job_runner_external_id + log.debug("Attempt remote Pulsar kill of job with url %s and id %s" % (pulsar_url, job_id)) + client = self.get_client(job.destination_params, job_id) + client.kill() + + def recover( self, job, job_wrapper ): + """Recovers jobs stuck in the queued/running state when Galaxy started""" + job_state = self._job_state( job, job_wrapper ) + job_wrapper.command_line = job.get_command_line() + state = job.get_state() + if state in [model.Job.states.RUNNING, model.Job.states.QUEUED]: + log.debug( "(Pulsar/%s) is still in running state, adding to the Pulsar queue" % ( job.get_id()) ) + job_state.old_state = True + job_state.running = state == model.Job.states.RUNNING + self.monitor_queue.put( job_state ) + + def shutdown( self ): + super( PulsarJobRunner, self ).shutdown() + self.client_manager.shutdown() + + def _job_state( self, job, job_wrapper ): + job_state = AsynchronousJobState() + # TODO: Determine why this is set when using normal message queue updates + # but not CLI submitted MQ updates... + raw_job_id = job.get_job_runner_external_id() or job_wrapper.job_id + job_state.job_id = str( raw_job_id ) + job_state.runner_url = job_wrapper.get_job_runner_url() + job_state.job_destination = job_wrapper.job_destination + job_state.job_wrapper = job_wrapper + return job_state + + def __client_outputs( self, client, job_wrapper ): + work_dir_outputs = self.get_work_dir_outputs( job_wrapper ) + output_files = self.get_output_files( job_wrapper ) + client_outputs = ClientOutputs( + working_directory=job_wrapper.working_directory, + work_dir_outputs=work_dir_outputs, + output_files=output_files, + version_file=job_wrapper.get_version_string_path(), + ) + return client_outputs + + @staticmethod + def __dependencies_description( pulsar_client, job_wrapper ): + dependency_resolution = PulsarJobRunner.__dependency_resolution( pulsar_client ) + remote_dependency_resolution = dependency_resolution == "remote" + if not remote_dependency_resolution: + return None + requirements = job_wrapper.tool.requirements or [] + installed_tool_dependencies = job_wrapper.tool.installed_tool_dependencies or [] + return dependencies.DependenciesDescription( + requirements=requirements, + installed_tool_dependencies=installed_tool_dependencies, + ) + + @staticmethod + def __dependency_resolution( pulsar_client ): + dependency_resolution = pulsar_client.destination_params.get( "dependency_resolution", "local" ) + if dependency_resolution not in ["none", "local", "remote"]: + raise Exception("Unknown dependency_resolution value encountered %s" % dependency_resolution) + return dependency_resolution + + @staticmethod + def __remote_metadata( pulsar_client ): + remote_metadata = string_as_bool_or_none( pulsar_client.destination_params.get( "remote_metadata", False ) ) + return remote_metadata + + @staticmethod + def __use_remote_datatypes_conf( pulsar_client ): + """ When setting remote metadata, use integrated datatypes from this + Galaxy instance or use the datatypes config configured via the remote + Pulsar. + + Both options are broken in different ways for same reason - datatypes + may not match. One can push the local datatypes config to the remote + server - but there is no guarentee these datatypes will be defined + there. Alternatively, one can use the remote datatype config - but + there is no guarentee that it will contain all the datatypes available + to this Galaxy. + """ + use_remote_datatypes = string_as_bool_or_none( pulsar_client.destination_params.get( "use_remote_datatypes", False ) ) + return use_remote_datatypes + + @staticmethod + def __rewrite_parameters( pulsar_client ): + return string_as_bool_or_none( pulsar_client.destination_params.get( "rewrite_parameters", False ) ) or False + + def __build_metadata_configuration(self, client, job_wrapper, remote_metadata, remote_job_config): + metadata_kwds = {} + if remote_metadata: + remote_system_properties = remote_job_config.get("system_properties", {}) + remote_galaxy_home = remote_system_properties.get("galaxy_home", None) + if not remote_galaxy_home: + raise Exception(NO_REMOTE_GALAXY_FOR_METADATA_MESSAGE) + metadata_kwds['exec_dir'] = remote_galaxy_home + outputs_directory = remote_job_config['outputs_directory'] + configs_directory = remote_job_config['configs_directory'] + working_directory = remote_job_config['working_directory'] + # For metadata calculation, we need to build a list of of output + # file objects with real path indicating location on Galaxy server + # and false path indicating location on compute server. Since the + # Pulsar disables from_work_dir copying as part of the job command + # line we need to take the list of output locations on the Pulsar + # server (produced by self.get_output_files(job_wrapper)) and for + # each work_dir output substitute the effective path on the Pulsar + # server relative to the remote working directory as the + # false_path to send the metadata command generation module. + work_dir_outputs = self.get_work_dir_outputs(job_wrapper, job_working_directory=working_directory) + outputs = [Bunch(false_path=os.path.join(outputs_directory, os.path.basename(path)), real_path=path) for path in self.get_output_files(job_wrapper)] + for output in outputs: + for pulsar_workdir_path, real_path in work_dir_outputs: + if real_path == output.real_path: + output.false_path = pulsar_workdir_path + metadata_kwds['output_fnames'] = outputs + metadata_kwds['compute_tmp_dir'] = working_directory + metadata_kwds['config_root'] = remote_galaxy_home + default_config_file = os.path.join(remote_galaxy_home, 'universe_wsgi.ini') + metadata_kwds['config_file'] = remote_system_properties.get('galaxy_config_file', default_config_file) + metadata_kwds['dataset_files_path'] = remote_system_properties.get('galaxy_dataset_files_path', None) + if PulsarJobRunner.__use_remote_datatypes_conf( client ): + remote_datatypes_config = remote_system_properties.get('galaxy_datatypes_config_file', None) + if not remote_datatypes_config: + log.warn(NO_REMOTE_DATATYPES_CONFIG) + remote_datatypes_config = os.path.join(remote_galaxy_home, 'datatypes_conf.xml') + metadata_kwds['datatypes_config'] = remote_datatypes_config + else: + integrates_datatypes_config = self.app.datatypes_registry.integrated_datatypes_configs + # Ensure this file gets pushed out to the remote config dir. + job_wrapper.extra_filenames.append(integrates_datatypes_config) + + metadata_kwds['datatypes_config'] = os.path.join(configs_directory, os.path.basename(integrates_datatypes_config)) + return metadata_kwds + + +class PulsarLegacyJobRunner( PulsarJobRunner ): + destination_defaults = dict( + rewrite_parameters="false", + dependency_resolution="local", + ) + + +class PulsarMQJobRunner( PulsarJobRunner ): + destination_defaults = dict( + default_file_action="remote_transfer", + rewrite_parameters="true", + dependency_resolution="remote", + jobs_directory=PARAMETER_SPECIFICATION_REQUIRED, + url=PARAMETER_SPECIFICATION_IGNORED, + private_token=PARAMETER_SPECIFICATION_IGNORED + ) + + def _monitor( self ): + # This is a message queue driven runner, don't monitor + # just setup required callback. + self.client_manager.ensure_has_status_update_callback(self.__async_update) + + def __async_update( self, full_status ): + job_id = None + try: + job_id = full_status[ "job_id" ] + job, job_wrapper = self.app.job_manager.job_handler.job_queue.job_pair_for_id( job_id ) + job_state = self._job_state( job, job_wrapper ) + self._update_job_state_for_status(job_state, full_status[ "status" ] ) + except Exception: + log.exception( "Failed to update Pulsar job status for job_id %s" % job_id ) + raise + # Nothing else to do? - Attempt to fail the job? + + +class PulsarRESTJobRunner( PulsarJobRunner ): + destination_defaults = dict( + default_file_action="transfer", + rewrite_parameters="true", + dependency_resolution="remote", + url=PARAMETER_SPECIFICATION_REQUIRED, + ) + + +class PulsarComputeEnvironment( ComputeEnvironment ): + + def __init__( self, pulsar_client, job_wrapper, remote_job_config ): + self.pulsar_client = pulsar_client + self.job_wrapper = job_wrapper + self.local_path_config = job_wrapper.default_compute_environment() + self.unstructured_path_rewrites = {} + # job_wrapper.prepare is going to expunge the job backing the following + # computations, so precalculate these paths. + self._wrapper_input_paths = self.local_path_config.input_paths() + self._wrapper_output_paths = self.local_path_config.output_paths() + self.path_mapper = PathMapper(pulsar_client, remote_job_config, self.local_path_config.working_directory()) + self._config_directory = remote_job_config[ "configs_directory" ] + self._working_directory = remote_job_config[ "working_directory" ] + self._sep = remote_job_config[ "system_properties" ][ "separator" ] + self._tool_dir = remote_job_config[ "tools_directory" ] + version_path = self.local_path_config.version_path() + new_version_path = self.path_mapper.remote_version_path_rewrite(version_path) + if new_version_path: + version_path = new_version_path + self._version_path = version_path + + def output_paths( self ): + local_output_paths = self._wrapper_output_paths + + results = [] + for local_output_path in local_output_paths: + wrapper_path = str( local_output_path ) + remote_path = self.path_mapper.remote_output_path_rewrite( wrapper_path ) + results.append( self._dataset_path( local_output_path, remote_path ) ) + return results + + def input_paths( self ): + local_input_paths = self._wrapper_input_paths + + results = [] + for local_input_path in local_input_paths: + wrapper_path = str( local_input_path ) + # This will over-copy in some cases. For instance in the case of task + # splitting, this input will be copied even though only the work dir + # input will actually be used. + remote_path = self.path_mapper.remote_input_path_rewrite( wrapper_path ) + results.append( self._dataset_path( local_input_path, remote_path ) ) + return results + + def _dataset_path( self, local_dataset_path, remote_path ): + remote_extra_files_path = None + if remote_path: + remote_extra_files_path = "%s_files" % remote_path[ 0:-len( ".dat" ) ] + return local_dataset_path.with_path_for_job( remote_path, remote_extra_files_path ) + + def working_directory( self ): + return self._working_directory + + def config_directory( self ): + return self._config_directory + + def new_file_path( self ): + return self.working_directory() # Problems with doing this? + + def sep( self ): + return self._sep + + def version_path( self ): + return self._version_path + + def rewriter( self, parameter_value ): + unstructured_path_rewrites = self.unstructured_path_rewrites + if parameter_value in unstructured_path_rewrites: + # Path previously mapped, use previous mapping. + return unstructured_path_rewrites[ parameter_value ] + if parameter_value in unstructured_path_rewrites.itervalues(): + # Path is a rewritten remote path (this might never occur, + # consider dropping check...) + return parameter_value + + rewrite, new_unstructured_path_rewrites = self.path_mapper.check_for_arbitrary_rewrite( parameter_value ) + if rewrite: + unstructured_path_rewrites.update(new_unstructured_path_rewrites) + return rewrite + else: + # Did need to rewrite, use original path or value. + return parameter_value + + def unstructured_path_rewriter( self ): + return self.rewriter diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/util/__init__.py --- a/lib/galaxy/jobs/runners/util/__init__.py +++ b/lib/galaxy/jobs/runners/util/__init__.py @@ -1,7 +1,7 @@ """ This module and its submodules contains utilities for running external processes and interfacing with job managers. This module should contain -functionality shared between Galaxy and the LWR. +functionality shared between Galaxy and the Pulsar. """ from galaxy.util.bunch import Bunch diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/util/cli/factory.py --- a/lib/galaxy/jobs/runners/util/cli/factory.py +++ b/lib/galaxy/jobs/runners/util/cli/factory.py @@ -5,7 +5,7 @@ ) code_dir = 'lib' except ImportError: - from lwr.managers.util.cli import ( + from pulsar.managers.util.cli import ( CliInterface, split_params ) diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/util/cli/job/slurm.py --- a/lib/galaxy/jobs/runners/util/cli/job/slurm.py +++ b/lib/galaxy/jobs/runners/util/cli/job/slurm.py @@ -5,7 +5,7 @@ from galaxy.model import Job job_states = Job.states except ImportError: - # Not in Galaxy, map Galaxy job states to LWR ones. + # Not in Galaxy, map Galaxy job states to Pulsar ones. from galaxy.util import enum job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/jobs/runners/util/cli/job/torque.py --- a/lib/galaxy/jobs/runners/util/cli/job/torque.py +++ b/lib/galaxy/jobs/runners/util/cli/job/torque.py @@ -7,7 +7,7 @@ from galaxy.model import Job job_states = Job.states except ImportError: - # Not in Galaxy, map Galaxy job states to LWR ones. + # Not in Galaxy, map Galaxy job states to Pulsar ones. from galaxy.util import enum job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/objectstore/__init__.py --- a/lib/galaxy/objectstore/__init__.py +++ b/lib/galaxy/objectstore/__init__.py @@ -623,9 +623,9 @@ elif store == 'irods': from .rods import IRODSObjectStore return IRODSObjectStore(config=config, config_xml=config_xml) - elif store == 'lwr': - from .lwr import LwrObjectStore - return LwrObjectStore(config=config, config_xml=config_xml) + elif store == 'pulsar': + from .pulsar import PulsarObjectStore + return PulsarObjectStore(config=config, config_xml=config_xml) else: log.error("Unrecognized object store definition: {0}".format(store)) diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/objectstore/lwr.py --- a/lib/galaxy/objectstore/lwr.py +++ /dev/null @@ -1,76 +0,0 @@ -from __future__ import absolute_import # Need to import lwr_client absolutely. -from ..objectstore import ObjectStore -try: - from galaxy.jobs.runners.lwr_client.manager import ObjectStoreClientManager -except ImportError: - from lwr.lwr_client.manager import ObjectStoreClientManager - - -class LwrObjectStore(ObjectStore): - """ - Object store implementation that delegates to a remote LWR server. - - This may be more aspirational than practical for now, it would be good to - Galaxy to a point that a handler thread could be setup that doesn't attempt - to access the disk files returned by a (this) object store - just passing - them along to the LWR unmodified. That modification - along with this - implementation and LWR job destinations would then allow Galaxy to fully - manage jobs on remote servers with completely different mount points. - - This implementation should be considered beta and may be dropped from - Galaxy at some future point or significantly modified. - """ - - def __init__(self, config, config_xml): - self.lwr_client = self.__build_lwr_client(config_xml) - - def exists(self, obj, **kwds): - return self.lwr_client.exists(**self.__build_kwds(obj, **kwds)) - - def file_ready(self, obj, **kwds): - return self.lwr_client.file_ready(**self.__build_kwds(obj, **kwds)) - - def create(self, obj, **kwds): - return self.lwr_client.create(**self.__build_kwds(obj, **kwds)) - - def empty(self, obj, **kwds): - return self.lwr_client.empty(**self.__build_kwds(obj, **kwds)) - - def size(self, obj, **kwds): - return self.lwr_client.size(**self.__build_kwds(obj, **kwds)) - - def delete(self, obj, **kwds): - return self.lwr_client.delete(**self.__build_kwds(obj, **kwds)) - - # TODO: Optimize get_data. - def get_data(self, obj, **kwds): - return self.lwr_client.get_data(**self.__build_kwds(obj, **kwds)) - - def get_filename(self, obj, **kwds): - return self.lwr_client.get_filename(**self.__build_kwds(obj, **kwds)) - - def update_from_file(self, obj, **kwds): - return self.lwr_client.update_from_file(**self.__build_kwds(obj, **kwds)) - - def get_store_usage_percent(self): - return self.lwr_client.get_store_usage_percent() - - def get_object_url(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None): - return None - - def __build_kwds(self, obj, **kwds): - kwds['object_id'] = obj.id - return kwds - pass - - def __build_lwr_client(self, config_xml): - url = config_xml.get("url") - private_token = config_xml.get("private_token", None) - transport = config_xml.get("transport", None) - manager_options = dict(transport=transport) - client_options = dict(url=url, private_token=private_token) - lwr_client = ObjectStoreClientManager(**manager_options).get_client(client_options) - return lwr_client - - def shutdown(self): - pass diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/objectstore/pulsar.py --- /dev/null +++ b/lib/galaxy/objectstore/pulsar.py @@ -0,0 +1,73 @@ +from __future__ import absolute_import # Need to import pulsar_client absolutely. +from ..objectstore import ObjectStore +from pulsar.client.manager import ObjectStoreClientManager + + +class PulsarObjectStore(ObjectStore): + """ + Object store implementation that delegates to a remote Pulsar server. + + This may be more aspirational than practical for now, it would be good to + Galaxy to a point that a handler thread could be setup that doesn't attempt + to access the disk files returned by a (this) object store - just passing + them along to the Pulsar unmodified. That modification - along with this + implementation and Pulsar job destinations would then allow Galaxy to fully + manage jobs on remote servers with completely different mount points. + + This implementation should be considered beta and may be dropped from + Galaxy at some future point or significantly modified. + """ + + def __init__(self, config, config_xml): + self.pulsar_client = self.__build_pulsar_client(config_xml) + + def exists(self, obj, **kwds): + return self.pulsar_client.exists(**self.__build_kwds(obj, **kwds)) + + def file_ready(self, obj, **kwds): + return self.pulsar_client.file_ready(**self.__build_kwds(obj, **kwds)) + + def create(self, obj, **kwds): + return self.pulsar_client.create(**self.__build_kwds(obj, **kwds)) + + def empty(self, obj, **kwds): + return self.pulsar_client.empty(**self.__build_kwds(obj, **kwds)) + + def size(self, obj, **kwds): + return self.pulsar_client.size(**self.__build_kwds(obj, **kwds)) + + def delete(self, obj, **kwds): + return self.pulsar_client.delete(**self.__build_kwds(obj, **kwds)) + + # TODO: Optimize get_data. + def get_data(self, obj, **kwds): + return self.pulsar_client.get_data(**self.__build_kwds(obj, **kwds)) + + def get_filename(self, obj, **kwds): + return self.pulsar_client.get_filename(**self.__build_kwds(obj, **kwds)) + + def update_from_file(self, obj, **kwds): + return self.pulsar_client.update_from_file(**self.__build_kwds(obj, **kwds)) + + def get_store_usage_percent(self): + return self.pulsar_client.get_store_usage_percent() + + def get_object_url(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None): + return None + + def __build_kwds(self, obj, **kwds): + kwds['object_id'] = obj.id + return kwds + pass + + def __build_pulsar_client(self, config_xml): + url = config_xml.get("url") + private_token = config_xml.get("private_token", None) + transport = config_xml.get("transport", None) + manager_options = dict(transport=transport) + client_options = dict(url=url, private_token=private_token) + pulsar_client = ObjectStoreClientManager(**manager_options).get_client(client_options) + return pulsar_client + + def shutdown(self): + pass diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/galaxy/tools/deps/dependencies.py --- a/lib/galaxy/tools/deps/dependencies.py +++ b/lib/galaxy/tools/deps/dependencies.py @@ -8,7 +8,7 @@ related context required to resolve dependencies via the ToolShedPackageDependencyResolver. - This is meant to enable remote resolution of dependencies, by the LWR or + This is meant to enable remote resolution of dependencies, by the Pulsar or other potential remote execution mechanisms. """ @@ -39,7 +39,7 @@ @staticmethod def _toolshed_install_dependency_from_dict(as_dict): - # Rather than requiring full models in LWR, just use simple objects + # Rather than requiring full models in Pulsar, just use simple objects # containing only properties and associations used to resolve # dependencies for tool execution. repository_object = bunch.Bunch( diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/pulsar/client/__init__.py --- /dev/null +++ b/lib/pulsar/client/__init__.py @@ -0,0 +1,62 @@ +""" +pulsar client +====== + +This module contains logic for interfacing with an external Pulsar server. + +------------------ +Configuring Galaxy +------------------ + +Galaxy job runners are configured in Galaxy's ``job_conf.xml`` file. See ``job_conf.xml.sample_advanced`` +in your Galaxy code base or on +`Bitbucket <https://bitbucket.org/galaxy/galaxy-dist/src/tip/job_conf.xml.sample_advanc…>`_ +for information on how to configure Galaxy to interact with the Pulsar. + +Galaxy also supports an older, less rich configuration of job runners directly +in its main ``universe_wsgi.ini`` file. The following section describes how to +configure Galaxy to communicate with the Pulsar in this legacy mode. + +Legacy +------ + +A Galaxy tool can be configured to be executed remotely via Pulsar by +adding a line to the ``universe_wsgi.ini`` file under the +``galaxy:tool_runners`` section with the format:: + + <tool_id> = pulsar://http://<pulsar_host>:<pulsar_port> + +As an example, if a host named remotehost is running the Pulsar server +application on port ``8913``, then the tool with id ``test_tool`` can +be configured to run remotely on remotehost by adding the following +line to ``universe.ini``:: + + test_tool = pulsar://http://remotehost:8913 + +Remember this must be added after the ``[galaxy:tool_runners]`` header +in the ``universe.ini`` file. + + +""" + +from .staging.down import finish_job +from .staging.up import submit_job +from .staging import ClientJobDescription +from .staging import PulsarOutputs +from .staging import ClientOutputs +from .client import OutputNotFoundException +from .manager import build_client_manager +from .destination import url_to_destination_params +from .path_mapper import PathMapper + +__all__ = [ + build_client_manager, + OutputNotFoundException, + url_to_destination_params, + finish_job, + submit_job, + ClientJobDescription, + PulsarOutputs, + ClientOutputs, + PathMapper, +] diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/pulsar/client/action_mapper.py --- /dev/null +++ b/lib/pulsar/client/action_mapper.py @@ -0,0 +1,567 @@ +from json import load +from os import makedirs +from os.path import exists +from os.path import abspath +from os.path import dirname +from os.path import join +from os.path import basename +from os.path import sep +import fnmatch +from re import compile +from re import escape +import galaxy.util +from galaxy.util.bunch import Bunch +from .config_util import read_file +from .util import directory_files +from .util import unique_path_prefix +from .transport import get_file +from .transport import post_file + + +DEFAULT_MAPPED_ACTION = 'transfer' # Not really clear to me what this should be, exception? +DEFAULT_PATH_MAPPER_TYPE = 'prefix' + +STAGING_ACTION_REMOTE = "remote" +STAGING_ACTION_LOCAL = "local" +STAGING_ACTION_NONE = None +STAGING_ACTION_DEFAULT = "default" + +# Poor man's enum. +path_type = Bunch( + # Galaxy input datasets and extra files. + INPUT="input", + # Galaxy config and param files. + CONFIG="config", + # Files from tool's tool_dir (for now just wrapper if available). + TOOL="tool", + # Input work dir files - e.g. metadata files, task-split input files, etc.. + WORKDIR="workdir", + # Galaxy output datasets in their final home. + OUTPUT="output", + # Galaxy from_work_dir output paths and other files (e.g. galaxy.json) + OUTPUT_WORKDIR="output_workdir", + # Other fixed tool parameter paths (likely coming from tool data, but not + # nessecarily). Not sure this is the best name... + UNSTRUCTURED="unstructured", +) + + +ACTION_DEFAULT_PATH_TYPES = [ + path_type.INPUT, + path_type.CONFIG, + path_type.TOOL, + path_type.WORKDIR, + path_type.OUTPUT, + path_type.OUTPUT_WORKDIR, +] +ALL_PATH_TYPES = ACTION_DEFAULT_PATH_TYPES + [path_type.UNSTRUCTURED] + + +class FileActionMapper(object): + """ + Objects of this class define how paths are mapped to actions. + + >>> json_string = r'''{"paths": [ \ + {"path": "/opt/galaxy", "action": "none"}, \ + {"path": "/galaxy/data", "action": "transfer"}, \ + {"path": "/cool/bamfiles/**/*.bam", "action": "copy", "match_type": "glob"}, \ + {"path": ".*/dataset_\\\\d+.dat", "action": "copy", "match_type": "regex"} \ + ]}''' + >>> from tempfile import NamedTemporaryFile + >>> from os import unlink + >>> def mapper_for(default_action, config_contents): + ... f = NamedTemporaryFile(delete=False) + ... f.write(config_contents.encode('UTF-8')) + ... f.close() + ... mock_client = Bunch(default_file_action=default_action, action_config_path=f.name, files_endpoint=None) + ... mapper = FileActionMapper(mock_client) + ... mapper = FileActionMapper(config=mapper.to_dict()) # Serialize and deserialize it to make sure still works + ... unlink(f.name) + ... return mapper + >>> mapper = mapper_for(default_action='none', config_contents=json_string) + >>> # Test first config line above, implicit path prefix mapper + >>> action = mapper.action('/opt/galaxy/tools/filters/catWrapper.py', 'input') + >>> action.action_type == u'none' + True + >>> action.staging_needed + False + >>> # Test another (2nd) mapper, this one with a different action + >>> action = mapper.action('/galaxy/data/files/000/dataset_1.dat', 'input') + >>> action.action_type == u'transfer' + True + >>> action.staging_needed + True + >>> # Always at least copy work_dir outputs. + >>> action = mapper.action('/opt/galaxy/database/working_directory/45.sh', 'workdir') + >>> action.action_type == u'copy' + True + >>> action.staging_needed + True + >>> # Test glob mapper (matching test) + >>> mapper.action('/cool/bamfiles/projectABC/study1/patient3.bam', 'input').action_type == u'copy' + True + >>> # Test glob mapper (non-matching test) + >>> mapper.action('/cool/bamfiles/projectABC/study1/patient3.bam.bai', 'input').action_type == u'none' + True + >>> # Regex mapper test. + >>> mapper.action('/old/galaxy/data/dataset_10245.dat', 'input').action_type == u'copy' + True + >>> # Doesn't map unstructured paths by default + >>> mapper.action('/old/galaxy/data/dataset_10245.dat', 'unstructured').action_type == u'none' + True + >>> input_only_mapper = mapper_for(default_action="none", config_contents=r'''{"paths": [ \ + {"path": "/", "action": "transfer", "path_types": "input"} \ + ] }''') + >>> input_only_mapper.action('/dataset_1.dat', 'input').action_type == u'transfer' + True + >>> input_only_mapper.action('/dataset_1.dat', 'output').action_type == u'none' + True + >>> unstructured_mapper = mapper_for(default_action="none", config_contents=r'''{"paths": [ \ + {"path": "/", "action": "transfer", "path_types": "*any*"} \ + ] }''') + >>> unstructured_mapper.action('/old/galaxy/data/dataset_10245.dat', 'unstructured').action_type == u'transfer' + True + """ + + def __init__(self, client=None, config=None): + if config is None and client is None: + message = "FileActionMapper must be constructed from either a client or a config dictionary." + raise Exception(message) + if config is None: + config = self.__client_to_config(client) + self.default_action = config.get("default_action", "transfer") + self.mappers = mappers_from_dicts(config.get("paths", [])) + self.files_endpoint = config.get("files_endpoint", None) + + def action(self, path, type, mapper=None): + mapper = self.__find_mapper(path, type, mapper) + action_class = self.__action_class(path, type, mapper) + file_lister = DEFAULT_FILE_LISTER + action_kwds = {} + if mapper: + file_lister = mapper.file_lister + action_kwds = mapper.action_kwds + action = action_class(path, file_lister=file_lister, **action_kwds) + self.__process_action(action, type) + return action + + def unstructured_mappers(self): + """ Return mappers that will map 'unstructured' files (i.e. go beyond + mapping inputs, outputs, and config files). + """ + return filter(lambda m: path_type.UNSTRUCTURED in m.path_types, self.mappers) + + def to_dict(self): + return dict( + default_action=self.default_action, + files_endpoint=self.files_endpoint, + paths=map(lambda m: m.to_dict(), self.mappers) + ) + + def __client_to_config(self, client): + action_config_path = client.action_config_path + if action_config_path: + config = read_file(action_config_path) + else: + config = dict() + config["default_action"] = client.default_file_action + config["files_endpoint"] = client.files_endpoint + return config + + def __load_action_config(self, path): + config = load(open(path, 'rb')) + self.mappers = mappers_from_dicts(config.get('paths', [])) + + def __find_mapper(self, path, type, mapper=None): + if not mapper: + normalized_path = abspath(path) + for query_mapper in self.mappers: + if query_mapper.matches(normalized_path, type): + mapper = query_mapper + break + return mapper + + def __action_class(self, path, type, mapper): + action_type = self.default_action if type in ACTION_DEFAULT_PATH_TYPES else "none" + if mapper: + action_type = mapper.action_type + if type in ["workdir", "output_workdir"] and action_type == "none": + # We are changing the working_directory relative to what + # Galaxy would use, these need to be copied over. + action_type = "copy" + action_class = actions.get(action_type, None) + if action_class is None: + message_template = "Unknown action_type encountered %s while trying to map path %s" + message_args = (action_type, path) + raise Exception(message_template % message_args) + return action_class + + def __process_action(self, action, file_type): + """ Extension point to populate extra action information after an + action has been created. + """ + if action.action_type == "remote_transfer": + url_base = self.files_endpoint + if not url_base: + raise Exception("Attempted to use remote_transfer action with defining a files_endpoint") + if "?" not in url_base: + url_base = "%s?" % url_base + # TODO: URL encode path. + url = "%s&path=%s&file_type=%s" % (url_base, action.path, file_type) + action.url = url + +REQUIRED_ACTION_KWD = object() + + +class BaseAction(object): + action_spec = {} + + def __init__(self, path, file_lister=None): + self.path = path + self.file_lister = file_lister or DEFAULT_FILE_LISTER + + def unstructured_map(self, path_helper): + unstructured_map = self.file_lister.unstructured_map(self.path) + if self.staging_needed: + # To ensure uniqueness, prepend unique prefix to each name + prefix = unique_path_prefix(self.path) + for path, name in unstructured_map.iteritems(): + unstructured_map[path] = join(prefix, name) + else: + path_rewrites = {} + for path in unstructured_map: + rewrite = self.path_rewrite(path_helper, path) + if rewrite: + path_rewrites[path] = rewrite + unstructured_map = path_rewrites + return unstructured_map + + @property + def staging_needed(self): + return self.staging != STAGING_ACTION_NONE + + @property + def staging_action_local(self): + return self.staging == STAGING_ACTION_LOCAL + + +class NoneAction(BaseAction): + """ This action indicates the corresponding path does not require any + additional action. This should indicate paths that are available both on + the Pulsar client (i.e. Galaxy server) and remote Pulsar server with the same + paths. """ + action_type = "none" + staging = STAGING_ACTION_NONE + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type) + + @classmethod + def from_dict(cls, action_dict): + return NoneAction(path=action_dict["path"]) + + def path_rewrite(self, path_helper, path=None): + return None + + +class RewriteAction(BaseAction): + """ This actin indicates the Pulsar server should simply rewrite the path + to the specified file. + """ + action_spec = dict( + source_directory=REQUIRED_ACTION_KWD, + destination_directory=REQUIRED_ACTION_KWD + ) + action_type = "rewrite" + staging = STAGING_ACTION_NONE + + def __init__(self, path, file_lister=None, source_directory=None, destination_directory=None): + self.path = path + self.file_lister = file_lister or DEFAULT_FILE_LISTER + self.source_directory = source_directory + self.destination_directory = destination_directory + + def to_dict(self): + return dict( + path=self.path, + action_type=self.action_type, + source_directory=self.source_directory, + destination_directory=self.destination_directory, + ) + + @classmethod + def from_dict(cls, action_dict): + return RewriteAction( + path=action_dict["path"], + source_directory=action_dict["source_directory"], + destination_directory=action_dict["destination_directory"], + ) + + def path_rewrite(self, path_helper, path=None): + if not path: + path = self.path + new_path = path_helper.from_posix_with_new_base(self.path, self.source_directory, self.destination_directory) + return None if new_path == self.path else new_path + + +class TransferAction(BaseAction): + """ This actions indicates that the Pulsar client should initiate an HTTP + transfer of the corresponding path to the remote Pulsar server before + launching the job. """ + action_type = "transfer" + staging = STAGING_ACTION_LOCAL + + +class CopyAction(BaseAction): + """ This action indicates that the Pulsar client should execute a file system + copy of the corresponding path to the Pulsar staging directory prior to + launching the corresponding job. """ + action_type = "copy" + staging = STAGING_ACTION_LOCAL + + +class RemoteCopyAction(BaseAction): + """ This action indicates the Pulsar server should copy the file before + execution via direct file system copy. This is like a CopyAction, but + it indicates the action should occur on the Pulsar server instead of on + the client. + """ + action_type = "remote_copy" + staging = STAGING_ACTION_REMOTE + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type) + + @classmethod + def from_dict(cls, action_dict): + return RemoteCopyAction(path=action_dict["path"]) + + def write_to_path(self, path): + galaxy.util.copy_to_path(open(self.path, "rb"), path) + + def write_from_path(self, pulsar_path): + destination = self.path + parent_directory = dirname(destination) + if not exists(parent_directory): + makedirs(parent_directory) + with open(pulsar_path, "rb") as f: + galaxy.util.copy_to_path(f, destination) + + +class RemoteTransferAction(BaseAction): + """ This action indicates the Pulsar server should copy the file before + execution via direct file system copy. This is like a CopyAction, but + it indicates the action should occur on the Pulsar server instead of on + the client. + """ + action_type = "remote_transfer" + staging = STAGING_ACTION_REMOTE + + def __init__(self, path, file_lister=None, url=None): + super(RemoteTransferAction, self).__init__(path, file_lister=file_lister) + self.url = url + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type, url=self.url) + + @classmethod + def from_dict(cls, action_dict): + return RemoteTransferAction(path=action_dict["path"], url=action_dict["url"]) + + def write_to_path(self, path): + get_file(self.url, path) + + def write_from_path(self, pulsar_path): + post_file(self.url, pulsar_path) + + +class MessageAction(object): + """ Sort of pseudo action describing "files" store in memory and + transferred via message (HTTP, Python-call, MQ, etc...) + """ + action_type = "message" + staging = STAGING_ACTION_DEFAULT + + def __init__(self, contents, client=None): + self.contents = contents + self.client = client + + @property + def staging_needed(self): + return True + + @property + def staging_action_local(self): + # Ekkk, cannot be called if created through from_dict. + # Shouldn't be a problem the way it is used - but is an + # object design problem. + return self.client.prefer_local_staging + + def to_dict(self): + return dict(contents=self.contents, action_type=MessageAction.action_type) + + @classmethod + def from_dict(cls, action_dict): + return MessageAction(contents=action_dict["contents"]) + + def write_to_path(self, path): + open(path, "w").write(self.contents) + + +DICTIFIABLE_ACTION_CLASSES = [RemoteCopyAction, RemoteTransferAction, MessageAction] + + +def from_dict(action_dict): + action_type = action_dict.get("action_type", None) + target_class = None + for action_class in DICTIFIABLE_ACTION_CLASSES: + if action_type == action_class.action_type: + target_class = action_class + if not target_class: + message = "Failed to recover action from dictionary - invalid action type specified %s." % action_type + raise Exception(message) + return target_class.from_dict(action_dict) + + +class BasePathMapper(object): + + def __init__(self, config): + action_type = config.get('action', DEFAULT_MAPPED_ACTION) + action_class = actions.get(action_type, None) + action_kwds = action_class.action_spec.copy() + for key, value in action_kwds.items(): + if key in config: + action_kwds[key] = config[key] + elif value is REQUIRED_ACTION_KWD: + message_template = "action_type %s requires key word argument %s" + message = message_template % (action_type, key) + raise Exception(message) + self.action_type = action_type + self.action_kwds = action_kwds + path_types_str = config.get('path_types', "*defaults*") + path_types_str = path_types_str.replace("*defaults*", ",".join(ACTION_DEFAULT_PATH_TYPES)) + path_types_str = path_types_str.replace("*any*", ",".join(ALL_PATH_TYPES)) + self.path_types = path_types_str.split(",") + self.file_lister = FileLister(config) + + def matches(self, path, path_type): + path_type_matches = path_type in self.path_types + return path_type_matches and self._path_matches(path) + + def _extend_base_dict(self, **kwds): + base_dict = dict( + action=self.action_type, + path_types=",".join(self.path_types), + match_type=self.match_type + ) + base_dict.update(self.file_lister.to_dict()) + base_dict.update(self.action_kwds) + base_dict.update(**kwds) + return base_dict + + +class PrefixPathMapper(BasePathMapper): + match_type = 'prefix' + + def __init__(self, config): + super(PrefixPathMapper, self).__init__(config) + self.prefix_path = abspath(config['path']) + + def _path_matches(self, path): + return path.startswith(self.prefix_path) + + def to_pattern(self): + pattern_str = "(%s%s[^\s,\"\']+)" % (escape(self.prefix_path), escape(sep)) + return compile(pattern_str) + + def to_dict(self): + return self._extend_base_dict(path=self.prefix_path) + + +class GlobPathMapper(BasePathMapper): + match_type = 'glob' + + def __init__(self, config): + super(GlobPathMapper, self).__init__(config) + self.glob_path = config['path'] + + def _path_matches(self, path): + return fnmatch.fnmatch(path, self.glob_path) + + def to_pattern(self): + return compile(fnmatch.translate(self.glob_path)) + + def to_dict(self): + return self._extend_base_dict(path=self.glob_path) + + +class RegexPathMapper(BasePathMapper): + match_type = 'regex' + + def __init__(self, config): + super(RegexPathMapper, self).__init__(config) + self.pattern_raw = config['path'] + self.pattern = compile(self.pattern_raw) + + def _path_matches(self, path): + return self.pattern.match(path) is not None + + def to_pattern(self): + return self.pattern + + def to_dict(self): + return self._extend_base_dict(path=self.pattern_raw) + +MAPPER_CLASSES = [PrefixPathMapper, GlobPathMapper, RegexPathMapper] +MAPPER_CLASS_DICT = dict(map(lambda c: (c.match_type, c), MAPPER_CLASSES)) + + +def mappers_from_dicts(mapper_def_list): + return map(lambda m: __mappper_from_dict(m), mapper_def_list) + + +def __mappper_from_dict(mapper_dict): + map_type = mapper_dict.get('match_type', DEFAULT_PATH_MAPPER_TYPE) + return MAPPER_CLASS_DICT[map_type](mapper_dict) + + +class FileLister(object): + + def __init__(self, config): + self.depth = int(config.get("depth", "0")) + + def to_dict(self): + return dict( + depth=self.depth + ) + + def unstructured_map(self, path): + depth = self.depth + if self.depth == 0: + return {path: basename(path)} + else: + while depth > 0: + path = dirname(path) + depth -= 1 + return dict([(join(path, f), f) for f in directory_files(path)]) + +DEFAULT_FILE_LISTER = FileLister(dict(depth=0)) + +ACTION_CLASSES = [ + NoneAction, + RewriteAction, + TransferAction, + CopyAction, + RemoteCopyAction, + RemoteTransferAction, +] +actions = dict([(clazz.action_type, clazz) for clazz in ACTION_CLASSES]) + + +__all__ = [ + FileActionMapper, + path_type, + from_dict, + MessageAction, + RemoteTransferAction, # For testing +] diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/pulsar/client/amqp_exchange.py --- /dev/null +++ b/lib/pulsar/client/amqp_exchange.py @@ -0,0 +1,139 @@ +try: + import kombu + from kombu import pools +except ImportError: + kombu = None + +import socket +import logging +import threading +from time import sleep +log = logging.getLogger(__name__) + + +KOMBU_UNAVAILABLE = "Attempting to bind to AMQP message queue, but kombu dependency unavailable" + +DEFAULT_EXCHANGE_NAME = "pulsar" +DEFAULT_EXCHANGE_TYPE = "direct" +# Set timeout to periodically give up looking and check if polling should end. +DEFAULT_TIMEOUT = 0.2 +DEFAULT_HEARTBEAT = 580 + +DEFAULT_RECONNECT_CONSUMER_WAIT = 1 +DEFAULT_HEARTBEAT_WAIT = 1 + + +class PulsarExchange(object): + """ Utility for publishing and consuming structured Pulsar queues using kombu. + This is shared between the server and client - an exchange should be setup + for each manager (or in the case of the client, each manager one wished to + communicate with.) + + Each Pulsar manager is defined solely by name in the scheme, so only one Pulsar + should target each AMQP endpoint or care should be taken that unique + manager names are used across Pulsar servers targetting same AMQP endpoint - + and in particular only one such Pulsar should define an default manager with + name _default_. + """ + + def __init__( + self, + url, + manager_name, + connect_ssl=None, + timeout=DEFAULT_TIMEOUT, + publish_kwds={}, + ): + """ + """ + if not kombu: + raise Exception(KOMBU_UNAVAILABLE) + self.__url = url + self.__manager_name = manager_name + self.__connect_ssl = connect_ssl + self.__exchange = kombu.Exchange(DEFAULT_EXCHANGE_NAME, DEFAULT_EXCHANGE_TYPE) + self.__timeout = timeout + # Be sure to log message publishing failures. + if publish_kwds.get("retry", False): + if "retry_policy" not in publish_kwds: + publish_kwds["retry_policy"] = {} + if "errback" not in publish_kwds["retry_policy"]: + publish_kwds["retry_policy"]["errback"] = self.__publish_errback + self.__publish_kwds = publish_kwds + + @property + def url(self): + return self.__url + + def consume(self, queue_name, callback, check=True, connection_kwargs={}): + queue = self.__queue(queue_name) + log.debug("Consuming queue '%s'", queue) + while check: + heartbeat_thread = None + try: + with self.connection(self.__url, heartbeat=DEFAULT_HEARTBEAT, **connection_kwargs) as connection: + with kombu.Consumer(connection, queues=[queue], callbacks=[callback], accept=['json']): + heartbeat_thread = self.__start_heartbeat(queue_name, connection) + while check and connection.connected: + try: + connection.drain_events(timeout=self.__timeout) + except socket.timeout: + pass + except (IOError, socket.error), exc: + # In testing, errno is None + log.warning('Got %s, will retry: %s', exc.__class__.__name__, exc) + if heartbeat_thread: + heartbeat_thread.join() + sleep(DEFAULT_RECONNECT_CONSUMER_WAIT) + + def heartbeat(self, connection): + log.debug('AMQP heartbeat thread alive') + while connection.connected: + connection.heartbeat_check() + sleep(DEFAULT_HEARTBEAT_WAIT) + log.debug('AMQP heartbeat thread exiting') + + def publish(self, name, payload): + with self.connection(self.__url) as connection: + with pools.producers[connection].acquire() as producer: + key = self.__queue_name(name) + producer.publish( + payload, + serializer='json', + exchange=self.__exchange, + declare=[self.__exchange], + routing_key=key, + **self.__publish_kwds + ) + + def __publish_errback(self, exc, interval): + log.error("Connection error while publishing: %r", exc, exc_info=1) + log.info("Retrying in %s seconds", interval) + + def connection(self, connection_string, **kwargs): + if "ssl" not in kwargs: + kwargs["ssl"] = self.__connect_ssl + return kombu.Connection(connection_string, **kwargs) + + def __queue(self, name): + queue_name = self.__queue_name(name) + queue = kombu.Queue(queue_name, self.__exchange, routing_key=queue_name) + return queue + + def __queue_name(self, name): + key_prefix = self.__key_prefix() + queue_name = '%s_%s' % (key_prefix, name) + return queue_name + + def __key_prefix(self): + if self.__manager_name == "_default_": + key_prefix = "pulsar_" + else: + key_prefix = "pulsar_%s_" % self.__manager_name + return key_prefix + + def __start_heartbeat(self, queue_name, connection): + thread_name = "consume-heartbeat-%s" % (self.__queue_name(queue_name)) + thread = threading.Thread(name=thread_name, target=self.heartbeat, args=(connection,)) + thread.start() + return thread diff -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee -r 566388d623749dad4259ea04cf44a5d858a56671 lib/pulsar/client/amqp_exchange_factory.py --- /dev/null +++ b/lib/pulsar/client/amqp_exchange_factory.py @@ -0,0 +1,41 @@ +from .amqp_exchange import PulsarExchange +from .util import filter_destination_params + + +def get_exchange(url, manager_name, params): + connect_ssl = parse_amqp_connect_ssl_params(params) + exchange_kwds = dict( + manager_name=manager_name, + connect_ssl=connect_ssl, + publish_kwds=parse_amqp_publish_kwds(params) + ) + timeout = params.get('amqp_consumer_timeout', False) + if timeout is not False: + exchange_kwds['timeout'] = timeout + exchange = PulsarExchange(url, **exchange_kwds) + return exchange + + +def parse_amqp_connect_ssl_params(params): + ssl_params = filter_destination_params(params, "amqp_connect_ssl_") + if not ssl_params: + return + + ssl = __import__('ssl') + if 'cert_reqs' in ssl_params: + value = ssl_params['cert_reqs'] + ssl_params['cert_reqs'] = getattr(ssl, value.upper()) + return ssl_params + + +def parse_amqp_publish_kwds(params): + all_publish_params = filter_destination_params(params, "amqp_publish_") + retry_policy_params = {} + for key in all_publish_params.keys(): + if key.startswith("retry_"): + value = all_publish_params[key] + retry_policy_params[key[len("retry_"):]] = value + del all_publish_params[key] + if retry_policy_params: + all_publish_params["retry_policy"] = retry_policy_params + return all_publish_params This diff is so big that we needed to truncate the remainder. Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: jmchilton: Merged in jmchilton/galaxy-central-fork-1 (pull request #411)
by commits-noreply＠bitbucket.org 23 Jun '14

23 Jun '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/4ef50975607e/ Changeset: 4ef50975607e User: jmchilton Date: 2014-06-24 04:15:44 Summary: Merged in jmchilton/galaxy-central-fork-1 (pull request #411) Allow dynamic job destinations rules to throw `galaxy.jobs.mapper.JobNotReadyException` indicating job not ready. Affected #: 2 files diff -r 5eaeb25bff8f383e54c36bd32e17432e6cafadde -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee lib/galaxy/jobs/handler.py --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -13,6 +13,7 @@ from galaxy import model from galaxy.util.sleeper import Sleeper from galaxy.jobs import JobWrapper, TaskWrapper, JobDestination +from galaxy.jobs.mapper import JobNotReadyException log = logging.getLogger( __name__ ) @@ -263,7 +264,7 @@ try: # Check the job's dependencies, requeue if they're not done. # Some of these states will only happen when using the in-memory job queue - job_state = self.__check_if_ready_to_run( job ) + job_state = self.__check_job_state( job ) if job_state == JOB_WAIT: new_waiting_jobs.append( job.id ) elif job_state == JOB_INPUT_ERROR: @@ -304,7 +305,7 @@ # Done with the session self.sa_session.remove() - def __check_if_ready_to_run( self, job ): + def __check_job_state( self, job ): """ Check if a job is ready to run by verifying that each of its input datasets is ready (specifically in the OK state). If any input dataset @@ -314,62 +315,97 @@ job can be dispatched. Otherwise, return JOB_WAIT indicating that input datasets are still being prepared. """ - # If tracking in the database, job.state is guaranteed to be NEW and the inputs are guaranteed to be OK if not self.track_jobs_in_database: - if job.state == model.Job.states.DELETED: - return JOB_DELETED - elif job.state == model.Job.states.ERROR: - return JOB_ADMIN_DELETED - for dataset_assoc in job.input_datasets + job.input_library_datasets: - idata = dataset_assoc.dataset - if not idata: - continue - # don't run jobs for which the input dataset was deleted - if idata.deleted: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) - return JOB_INPUT_DELETED - # an error in the input data causes us to bail immediately - elif idata.state == idata.states.ERROR: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state == idata.states.FAILED_METADATA: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): - # need to requeue - return JOB_WAIT + in_memory_not_ready_state = self.__verify_in_memory_job_inputs( job ) + if in_memory_not_ready_state: + return in_memory_not_ready_state + + # Else, if tracking in the database, job.state is guaranteed to be NEW and + # the inputs are guaranteed to be OK. + # Create the job wrapper so that the destination can be set - if job.id not in self.job_wrappers: - self.job_wrappers[job.id] = self.job_wrapper( job ) - # Cause the job_destination to be set and cached by the mapper + job_id = job.id + job_wrapper = self.job_wrappers.get( job_id, None ) + if not job_wrapper: + job_wrapper = self.job_wrapper( job ) + self.job_wrappers[ job_id ] = job_wrapper + + # If state == JOB_READY, assume job_destination also set - otherwise + # in case of various error or cancelled states do not assume + # destination has been set. + state, job_destination = self.__verify_job_ready( job, job_wrapper ) + + if state == JOB_READY: + # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration + self.increase_running_job_count(job.user_id, job_destination.id ) + return state + + def __verify_job_ready( self, job, job_wrapper ): + """ Compute job destination and verify job is ready at that + destination by checking job limits and quota. If this method + return a job state of JOB_READY - it MUST also return a job + destination. + """ + job_destination = None try: - self.job_wrappers[job.id].job_destination + # Cause the job_destination to be set and cached by the mapper + job_destination = job_wrapper.job_destination + except JobNotReadyException as e: + job_state = e.job_state or JOB_WAIT + return job_state, None except Exception, e: failure_message = getattr( e, 'failure_message', DEFAULT_JOB_PUT_FAILURE_MESSAGE ) if failure_message == DEFAULT_JOB_PUT_FAILURE_MESSAGE: log.exception( 'Failed to generate job destination' ) else: log.debug( "Intentionally failing job with message (%s)" % failure_message ) - self.job_wrappers[job.id].fail( failure_message ) - return JOB_ERROR + job_wrapper.fail( failure_message ) + return JOB_ERROR, job_destination # job is ready to run, check limits # TODO: these checks should be refactored to minimize duplication and made more modular/pluggable - state = self.__check_destination_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_destination_jobs( job, job_wrapper ) if state == JOB_READY: - state = self.__check_user_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_user_jobs( job, job_wrapper ) if state == JOB_READY and self.app.config.enable_quotas: quota = self.app.quota_agent.get_quota( job.user ) if quota is not None: try: usage = self.app.quota_agent.get_usage( user=job.user, history=job.history ) if usage > quota: - return JOB_USER_OVER_QUOTA + return JOB_USER_OVER_QUOTA, job_destination except AssertionError, e: pass # No history, should not happen with an anon user - if state == JOB_READY: - # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration - self.increase_running_job_count(job.user_id, self.job_wrappers[job.id].job_destination.id) - return state + return state, job_destination + + def __verify_in_memory_job_inputs( self, job ): + """ Perform the same checks that happen via SQL for in-memory managed + jobs. + """ + if job.state == model.Job.states.DELETED: + return JOB_DELETED + elif job.state == model.Job.states.ERROR: + return JOB_ADMIN_DELETED + for dataset_assoc in job.input_datasets + job.input_library_datasets: + idata = dataset_assoc.dataset + if not idata: + continue + # don't run jobs for which the input dataset was deleted + if idata.deleted: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) + return JOB_INPUT_DELETED + # an error in the input data causes us to bail immediately + elif idata.state == idata.states.ERROR: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state == idata.states.FAILED_METADATA: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): + # need to requeue + return JOB_WAIT + + # All inputs ready to go. + return None def __clear_job_count( self ): self.user_job_count = None diff -r 5eaeb25bff8f383e54c36bd32e17432e6cafadde -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee lib/galaxy/jobs/mapper.py --- a/lib/galaxy/jobs/mapper.py +++ b/lib/galaxy/jobs/mapper.py @@ -16,6 +16,13 @@ self.failure_message = failure_message +class JobNotReadyException( Exception ): + + def __init__( self, job_state=None, message=None ): + self.job_state = job_state + self.message = message + + class JobRunnerMapper( object ): """ This class is responsible to managing the mapping of jobs Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 5 new changesets
by commits-noreply＠bitbucket.org 23 Jun '14

23 Jun '14

5 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/72a0157988f1/ Changeset: 72a0157988f1 User: jmchilton Date: 2014-06-11 07:19:20 Summary: Refactor to simplify __check_if_ready_to_run. Move checks for in memory managed jobs prior to common path through code into their own helper method. Affected #: 1 file diff -r c52cb1f827d49f0f0ee2c1c97d3626dd1b2cc348 -r 72a0157988f1cf7e9440dc1cb3dc13ca6cdc42ba lib/galaxy/jobs/handler.py --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -240,7 +240,7 @@ try: # Check the job's dependencies, requeue if they're not done. # Some of these states will only happen when using the in-memory job queue - job_state = self.__check_if_ready_to_run( job ) + job_state = self.__check_job_state( job ) if job_state == JOB_WAIT: new_waiting_jobs.append( job.id ) elif job_state == JOB_INPUT_ERROR: @@ -281,7 +281,7 @@ # Done with the session self.sa_session.remove() - def __check_if_ready_to_run( self, job ): + def __check_job_state( self, job ): """ Check if a job is ready to run by verifying that each of its input datasets is ready (specifically in the OK state). If any input dataset @@ -293,28 +293,12 @@ """ # If tracking in the database, job.state is guaranteed to be NEW and the inputs are guaranteed to be OK if not self.track_jobs_in_database: - if job.state == model.Job.states.DELETED: - return JOB_DELETED - elif job.state == model.Job.states.ERROR: - return JOB_ADMIN_DELETED - for dataset_assoc in job.input_datasets + job.input_library_datasets: - idata = dataset_assoc.dataset - if not idata: - continue - # don't run jobs for which the input dataset was deleted - if idata.deleted: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) - return JOB_INPUT_DELETED - # an error in the input data causes us to bail immediately - elif idata.state == idata.states.ERROR: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state == idata.states.FAILED_METADATA: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): - # need to requeue - return JOB_WAIT + in_memory_job_state = self.__check_in_memory_job_state( job ) + if in_memory_job_state: + return in_memory_job_state + # If no problems, all inputs ready to go, continue + # with same checks either way. + # Create the job wrapper so that the destination can be set if job.id not in self.job_wrappers: self.job_wrappers[job.id] = self.job_wrapper( job ) @@ -348,6 +332,33 @@ self.increase_running_job_count(job.user_id, self.job_wrappers[job.id].job_destination.id) return state + def __check_in_memory_job_state( self, job ): + if job.state == model.Job.states.DELETED: + return JOB_DELETED + elif job.state == model.Job.states.ERROR: + return JOB_ADMIN_DELETED + for dataset_assoc in job.input_datasets + job.input_library_datasets: + idata = dataset_assoc.dataset + if not idata: + continue + # don't run jobs for which the input dataset was deleted + if idata.deleted: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) + return JOB_INPUT_DELETED + # an error in the input data causes us to bail immediately + elif idata.state == idata.states.ERROR: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state == idata.states.FAILED_METADATA: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): + # need to requeue + return JOB_WAIT + + # All inputs ready to go. + return None + def __clear_job_count( self ): self.user_job_count = None self.user_job_count_per_destination = None https://bitbucket.org/galaxy/galaxy-central/commits/519050ab95d7/ Changeset: 519050ab95d7 User: jmchilton Date: 2014-06-11 07:19:20 Summary: Refactor __check_if_ready_to_run - less code duplication... ... in the sense that it is no longer repeatedly grabbing same values from map, from wrapper, etc. Affected #: 1 file diff -r 72a0157988f1cf7e9440dc1cb3dc13ca6cdc42ba -r 519050ab95d7177abdea8d55bae3b03a3a289476 lib/galaxy/jobs/handler.py --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -291,33 +291,37 @@ job can be dispatched. Otherwise, return JOB_WAIT indicating that input datasets are still being prepared. """ - # If tracking in the database, job.state is guaranteed to be NEW and the inputs are guaranteed to be OK if not self.track_jobs_in_database: - in_memory_job_state = self.__check_in_memory_job_state( job ) - if in_memory_job_state: - return in_memory_job_state - # If no problems, all inputs ready to go, continue - # with same checks either way. + in_memory_not_ready_state = self.__verify_in_memory_job_inputs( job ) + if in_memory_not_ready_state: + return in_memory_not_ready_state + + # Else, if tracking in the database, job.state is guaranteed to be NEW and + # the inputs are guaranteed to be OK. # Create the job wrapper so that the destination can be set - if job.id not in self.job_wrappers: - self.job_wrappers[job.id] = self.job_wrapper( job ) + job_id = job.id + job_wrapper = self.job_wrappers.get( job_id, None ) + if not job_wrapper: + job_wrapper = self.job_wrapper( job ) + self.job_wrappers[ job_id ] = job_wrapper + # Cause the job_destination to be set and cached by the mapper try: - self.job_wrappers[job.id].job_destination + job_destination = job_wrapper.job_destination except Exception, e: failure_message = getattr( e, 'failure_message', DEFAULT_JOB_PUT_FAILURE_MESSAGE ) if failure_message == DEFAULT_JOB_PUT_FAILURE_MESSAGE: log.exception( 'Failed to generate job destination' ) else: log.debug( "Intentionally failing job with message (%s)" % failure_message ) - self.job_wrappers[job.id].fail( failure_message ) + job_wrapper.fail( failure_message ) return JOB_ERROR # job is ready to run, check limits # TODO: these checks should be refactored to minimize duplication and made more modular/pluggable - state = self.__check_destination_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_destination_jobs( job, job_wrapper ) if state == JOB_READY: - state = self.__check_user_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_user_jobs( job, job_wrapper ) if state == JOB_READY and self.app.config.enable_quotas: quota = self.app.quota_agent.get_quota( job.user ) if quota is not None: @@ -329,10 +333,13 @@ pass # No history, should not happen with an anon user if state == JOB_READY: # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration - self.increase_running_job_count(job.user_id, self.job_wrappers[job.id].job_destination.id) + self.increase_running_job_count(job.user_id, job_destination.id ) return state - def __check_in_memory_job_state( self, job ): + def __verify_in_memory_job_inputs( self, job ): + """ Perform the same checks that happen via SQL for in-memory managed + jobs. + """ if job.state == model.Job.states.DELETED: return JOB_DELETED elif job.state == model.Job.states.ERROR: https://bitbucket.org/galaxy/galaxy-central/commits/8bbb244d339b/ Changeset: 8bbb244d339b User: jmchilton Date: 2014-06-11 07:19:20 Summary: Break up __check_job_state - extract new __verify_job_ready method. Affected #: 1 file diff -r 519050ab95d7177abdea8d55bae3b03a3a289476 -r 8bbb244d339b396c841341faaa9353f4a1c5e5f1 lib/galaxy/jobs/handler.py --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -306,8 +306,25 @@ job_wrapper = self.job_wrapper( job ) self.job_wrappers[ job_id ] = job_wrapper - # Cause the job_destination to be set and cached by the mapper + # If state == JOB_READY, assume job_destination also set - otherwise + # in case of various error or cancelled states do not assume + # destination has been set. + state, job_destination = self.__verify_job_ready( job, job_wrapper ) + + if state == JOB_READY: + # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration + self.increase_running_job_count(job.user_id, job_destination.id ) + return state + + def __verify_job_ready( self, job, job_wrapper ): + """ Compute job destination and verify job is ready at that + destination by checking job limits and quota. If this method + return a job state of JOB_READY - it MUST also return a job + destination. + """ + job_destination = None try: + # Cause the job_destination to be set and cached by the mapper job_destination = job_wrapper.job_destination except Exception, e: failure_message = getattr( e, 'failure_message', DEFAULT_JOB_PUT_FAILURE_MESSAGE ) @@ -316,7 +333,7 @@ else: log.debug( "Intentionally failing job with message (%s)" % failure_message ) job_wrapper.fail( failure_message ) - return JOB_ERROR + return JOB_ERROR, job_destination # job is ready to run, check limits # TODO: these checks should be refactored to minimize duplication and made more modular/pluggable state = self.__check_destination_jobs( job, job_wrapper ) @@ -328,13 +345,10 @@ try: usage = self.app.quota_agent.get_usage( user=job.user, history=job.history ) if usage > quota: - return JOB_USER_OVER_QUOTA + return JOB_USER_OVER_QUOTA, job_destination except AssertionError, e: pass # No history, should not happen with an anon user - if state == JOB_READY: - # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration - self.increase_running_job_count(job.user_id, job_destination.id ) - return state + return state, job_destination def __verify_in_memory_job_inputs( self, job ): """ Perform the same checks that happen via SQL for in-memory managed https://bitbucket.org/galaxy/galaxy-central/commits/e1c8d892be5d/ Changeset: e1c8d892be5d User: jmchilton Date: 2014-06-11 07:19:20 Summary: Allow dynamic job destinations to throw new JobNotReadyException to indicate delayed decision. Perhaps wait on some deployer defined limit or resource to become available, etc... Affected #: 2 files diff -r 8bbb244d339b396c841341faaa9353f4a1c5e5f1 -r e1c8d892be5de20ed4564a3e467943701dccacd8 lib/galaxy/jobs/handler.py --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -13,6 +13,7 @@ from galaxy import model from galaxy.util.sleeper import Sleeper from galaxy.jobs import JobWrapper, TaskWrapper, JobDestination +from galaxy.jobs.mapper import JobNotReadyException log = logging.getLogger( __name__ ) @@ -326,6 +327,9 @@ try: # Cause the job_destination to be set and cached by the mapper job_destination = job_wrapper.job_destination + except JobNotReadyException as e: + job_state = e.job_state or JOB_WAIT + return job_state, None except Exception, e: failure_message = getattr( e, 'failure_message', DEFAULT_JOB_PUT_FAILURE_MESSAGE ) if failure_message == DEFAULT_JOB_PUT_FAILURE_MESSAGE: diff -r 8bbb244d339b396c841341faaa9353f4a1c5e5f1 -r e1c8d892be5de20ed4564a3e467943701dccacd8 lib/galaxy/jobs/mapper.py --- a/lib/galaxy/jobs/mapper.py +++ b/lib/galaxy/jobs/mapper.py @@ -16,6 +16,13 @@ self.failure_message = failure_message +class JobNotReadyException( Exception ): + + def __init__( self, job_state=None, message=None ): + self.job_state = job_state + self.message = message + + class JobRunnerMapper( object ): """ This class is responsible to managing the mapping of jobs https://bitbucket.org/galaxy/galaxy-central/commits/4ef50975607e/ Changeset: 4ef50975607e User: jmchilton Date: 2014-06-24 04:15:44 Summary: Merged in jmchilton/galaxy-central-fork-1 (pull request #411) Allow dynamic job destinations rules to throw `galaxy.jobs.mapper.JobNotReadyException` indicating job not ready. Affected #: 2 files diff -r 5eaeb25bff8f383e54c36bd32e17432e6cafadde -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee lib/galaxy/jobs/handler.py --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -13,6 +13,7 @@ from galaxy import model from galaxy.util.sleeper import Sleeper from galaxy.jobs import JobWrapper, TaskWrapper, JobDestination +from galaxy.jobs.mapper import JobNotReadyException log = logging.getLogger( __name__ ) @@ -263,7 +264,7 @@ try: # Check the job's dependencies, requeue if they're not done. # Some of these states will only happen when using the in-memory job queue - job_state = self.__check_if_ready_to_run( job ) + job_state = self.__check_job_state( job ) if job_state == JOB_WAIT: new_waiting_jobs.append( job.id ) elif job_state == JOB_INPUT_ERROR: @@ -304,7 +305,7 @@ # Done with the session self.sa_session.remove() - def __check_if_ready_to_run( self, job ): + def __check_job_state( self, job ): """ Check if a job is ready to run by verifying that each of its input datasets is ready (specifically in the OK state). If any input dataset @@ -314,62 +315,97 @@ job can be dispatched. Otherwise, return JOB_WAIT indicating that input datasets are still being prepared. """ - # If tracking in the database, job.state is guaranteed to be NEW and the inputs are guaranteed to be OK if not self.track_jobs_in_database: - if job.state == model.Job.states.DELETED: - return JOB_DELETED - elif job.state == model.Job.states.ERROR: - return JOB_ADMIN_DELETED - for dataset_assoc in job.input_datasets + job.input_library_datasets: - idata = dataset_assoc.dataset - if not idata: - continue - # don't run jobs for which the input dataset was deleted - if idata.deleted: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) - return JOB_INPUT_DELETED - # an error in the input data causes us to bail immediately - elif idata.state == idata.states.ERROR: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state == idata.states.FAILED_METADATA: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): - # need to requeue - return JOB_WAIT + in_memory_not_ready_state = self.__verify_in_memory_job_inputs( job ) + if in_memory_not_ready_state: + return in_memory_not_ready_state + + # Else, if tracking in the database, job.state is guaranteed to be NEW and + # the inputs are guaranteed to be OK. + # Create the job wrapper so that the destination can be set - if job.id not in self.job_wrappers: - self.job_wrappers[job.id] = self.job_wrapper( job ) - # Cause the job_destination to be set and cached by the mapper + job_id = job.id + job_wrapper = self.job_wrappers.get( job_id, None ) + if not job_wrapper: + job_wrapper = self.job_wrapper( job ) + self.job_wrappers[ job_id ] = job_wrapper + + # If state == JOB_READY, assume job_destination also set - otherwise + # in case of various error or cancelled states do not assume + # destination has been set. + state, job_destination = self.__verify_job_ready( job, job_wrapper ) + + if state == JOB_READY: + # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration + self.increase_running_job_count(job.user_id, job_destination.id ) + return state + + def __verify_job_ready( self, job, job_wrapper ): + """ Compute job destination and verify job is ready at that + destination by checking job limits and quota. If this method + return a job state of JOB_READY - it MUST also return a job + destination. + """ + job_destination = None try: - self.job_wrappers[job.id].job_destination + # Cause the job_destination to be set and cached by the mapper + job_destination = job_wrapper.job_destination + except JobNotReadyException as e: + job_state = e.job_state or JOB_WAIT + return job_state, None except Exception, e: failure_message = getattr( e, 'failure_message', DEFAULT_JOB_PUT_FAILURE_MESSAGE ) if failure_message == DEFAULT_JOB_PUT_FAILURE_MESSAGE: log.exception( 'Failed to generate job destination' ) else: log.debug( "Intentionally failing job with message (%s)" % failure_message ) - self.job_wrappers[job.id].fail( failure_message ) - return JOB_ERROR + job_wrapper.fail( failure_message ) + return JOB_ERROR, job_destination # job is ready to run, check limits # TODO: these checks should be refactored to minimize duplication and made more modular/pluggable - state = self.__check_destination_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_destination_jobs( job, job_wrapper ) if state == JOB_READY: - state = self.__check_user_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_user_jobs( job, job_wrapper ) if state == JOB_READY and self.app.config.enable_quotas: quota = self.app.quota_agent.get_quota( job.user ) if quota is not None: try: usage = self.app.quota_agent.get_usage( user=job.user, history=job.history ) if usage > quota: - return JOB_USER_OVER_QUOTA + return JOB_USER_OVER_QUOTA, job_destination except AssertionError, e: pass # No history, should not happen with an anon user - if state == JOB_READY: - # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration - self.increase_running_job_count(job.user_id, self.job_wrappers[job.id].job_destination.id) - return state + return state, job_destination + + def __verify_in_memory_job_inputs( self, job ): + """ Perform the same checks that happen via SQL for in-memory managed + jobs. + """ + if job.state == model.Job.states.DELETED: + return JOB_DELETED + elif job.state == model.Job.states.ERROR: + return JOB_ADMIN_DELETED + for dataset_assoc in job.input_datasets + job.input_library_datasets: + idata = dataset_assoc.dataset + if not idata: + continue + # don't run jobs for which the input dataset was deleted + if idata.deleted: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) + return JOB_INPUT_DELETED + # an error in the input data causes us to bail immediately + elif idata.state == idata.states.ERROR: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state == idata.states.FAILED_METADATA: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): + # need to requeue + return JOB_WAIT + + # All inputs ready to go. + return None def __clear_job_count( self ): self.user_job_count = None diff -r 5eaeb25bff8f383e54c36bd32e17432e6cafadde -r 4ef50975607ee0d3b4490c9e7a5795d6a04802ee lib/galaxy/jobs/mapper.py --- a/lib/galaxy/jobs/mapper.py +++ b/lib/galaxy/jobs/mapper.py @@ -16,6 +16,13 @@ self.failure_message = failure_message +class JobNotReadyException( Exception ): + + def __init__( self, job_state=None, message=None ): + self.job_state = job_state + self.message = message + + class JobRunnerMapper( object ): """ This class is responsible to managing the mapping of jobs Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: jmchilton: Docfix: Delete diminutive docker doc deficiency.
by commits-noreply＠bitbucket.org 23 Jun '14

23 Jun '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/5eaeb25bff8f/ Changeset: 5eaeb25bff8f User: jmchilton Date: 2014-06-23 19:49:05 Summary: Docfix: Delete diminutive docker doc deficiency. Affected #: 1 file diff -r 631127bb6c2657140e75f203301abfdf567d82e5 -r 5eaeb25bff8f383e54c36bd32e17432e6cafadde job_conf.xml.sample_advanced --- a/job_conf.xml.sample_advanced +++ b/job_conf.xml.sample_advanced @@ -178,7 +178,7 @@ does trust tool's specified container - but also wants tool's not configured to run in a container the following option can provide a fallback. --> -  + </destination><destination id="pbs" runner="pbs" tags="mycluster"/> Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: greg: Missed committing a new __init__.py file for Tool Shed dependencies.
by commits-noreply＠bitbucket.org 23 Jun '14

23 Jun '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/631127bb6c26/ Changeset: 631127bb6c26 User: greg Date: 2014-06-23 14:58:25 Summary: Missed committing a new __init__.py file for Tool Shed dependencies. Affected #: 1 file Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: greg: A bit of repository dependency function re-factoring for Galaxy installs.
by commits-noreply＠bitbucket.org 22 Jun '14

22 Jun '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/f5589d243d0f/ Changeset: f5589d243d0f User: greg Date: 2014-06-23 02:31:19 Summary: A bit of repository dependency function re-factoring for Galaxy installs. Affected #: 4 files diff -r 344e84a6554ae94604f32e8f62feebd65a6ac358 -r f5589d243d0f920cdd96c3e980762bb1a7cc061b lib/galaxy/webapps/galaxy/controllers/admin_toolshed.py --- a/lib/galaxy/webapps/galaxy/controllers/admin_toolshed.py +++ b/lib/galaxy/webapps/galaxy/controllers/admin_toolshed.py @@ -1359,9 +1359,8 @@ else: # Entering this else block occurs only if the tool_shed_repository does not include any valid tools. if install_repository_dependencies: - repository_dependencies = \ - repository_dependency_util.get_repository_dependencies_for_installed_tool_shed_repository( trans.app, - tool_shed_repository ) + repository_dependencies = rdm.get_repository_dependencies_for_installed_tool_shed_repository( trans.app, + tool_shed_repository ) else: repository_dependencies = None if metadata: @@ -1579,9 +1578,9 @@ raw_text = common_util.tool_shed_get( trans.app, tool_shed_url, url ) readme_files_dict = json.from_json_string( raw_text ) tool_dependencies = metadata.get( 'tool_dependencies', None ) - repository_dependencies = \ - repository_dependency_util.get_repository_dependencies_for_installed_tool_shed_repository( trans.app, - tool_shed_repository ) + rdm = RepositoryDependencyManager( trans.app ) + repository_dependencies = rdm.get_repository_dependencies_for_installed_tool_shed_repository( trans.app, + tool_shed_repository ) repo_info_dict = \ repository_maintenance_util.create_repo_info_dict( app=trans.app, repository_clone_url=repository_clone_url, diff -r 344e84a6554ae94604f32e8f62feebd65a6ac358 -r f5589d243d0f920cdd96c3e980762bb1a7cc061b lib/tool_shed/galaxy_install/repair_repository_manager.py --- a/lib/tool_shed/galaxy_install/repair_repository_manager.py +++ b/lib/tool_shed/galaxy_install/repair_repository_manager.py @@ -4,6 +4,7 @@ log = logging.getLogger( __name__ ) from tool_shed.galaxy_install import install_manager +from tool_shed.galaxy_install.repository_dependencies.repository_dependency_manager import RepositoryDependencyManager from tool_shed.util import common_util from tool_shed.util import container_util @@ -55,6 +56,7 @@ issues with an installed repository that has installation problems somewhere in its dependency hierarchy. """ + rdm = RepositoryDependencyManager( self.app ) tsr_ids = [] repo_info_dicts = [] tool_panel_section_keys = [] @@ -62,8 +64,8 @@ irm = install_manager.InstallRepositoryManager( self.app ) # Get a dictionary of all repositories upon which the contents of the current repository_metadata #record depend. - repository_dependencies_dict = \ - repository_dependency_util.get_repository_dependencies_for_installed_tool_shed_repository( self.app, repository ) + repository_dependencies_dict = rdm.get_repository_dependencies_for_installed_tool_shed_repository( self.app, + repository ) if repository_dependencies_dict: # Generate the list of installed repositories from the information contained in the # repository_dependencies dictionary. @@ -73,13 +75,15 @@ # repaired in the required order. for installed_repository in installed_repositories: tsr_ids.append( self.app.security.encode_id( installed_repository.id ) ) - repo_info_dict, tool_panel_section_key = self.get_repo_info_dict_for_repair( installed_repository ) + repo_info_dict, tool_panel_section_key = self.get_repo_info_dict_for_repair( rdm, + installed_repository ) tool_panel_section_keys.append( tool_panel_section_key ) repo_info_dicts.append( repo_info_dict ) else: # The received repository has no repository dependencies. tsr_ids.append( self.app.security.encode_id( repository.id ) ) - repo_info_dict, tool_panel_section_key = self.get_repo_info_dict_for_repair( repository ) + repo_info_dict, tool_panel_section_key = self.get_repo_info_dict_for_repair( rdm, + repository ) tool_panel_section_keys.append( tool_panel_section_key ) repo_info_dicts.append( repo_info_dict ) ordered_tsr_ids, ordered_repo_info_dicts, ordered_tool_panel_section_keys = \ @@ -91,11 +95,11 @@ repair_dict[ 'ordered_tool_panel_section_keys' ] = ordered_tool_panel_section_keys return repair_dict - def get_repo_info_dict_for_repair( self, repository ): + def get_repo_info_dict_for_repair( self, rdm, repository ): tool_panel_section_key = None repository_clone_url = common_util.generate_clone_url_for_installed_repository( self.app, repository ) - repository_dependencies = \ - repository_dependency_util.get_repository_dependencies_for_installed_tool_shed_repository( self.app, repository ) + repository_dependencies = rdm.get_repository_dependencies_for_installed_tool_shed_repository( self.app, + repository ) metadata = repository.metadata if metadata: tool_dependencies = metadata.get( 'tool_dependencies', None ) diff -r 344e84a6554ae94604f32e8f62feebd65a6ac358 -r f5589d243d0f920cdd96c3e980762bb1a7cc061b lib/tool_shed/galaxy_install/repository_dependencies/repository_dependency_manager.py --- a/lib/tool_shed/galaxy_install/repository_dependencies/repository_dependency_manager.py +++ b/lib/tool_shed/galaxy_install/repository_dependencies/repository_dependency_manager.py @@ -255,6 +255,29 @@ self.build_repository_dependency_relationships( all_repo_info_dicts, all_created_or_updated_tool_shed_repositories ) return created_or_updated_tool_shed_repositories, tool_panel_section_keys, all_repo_info_dicts, filtered_repo_info_dicts + def get_repository_dependencies_for_installed_tool_shed_repository( self, app, repository ): + """ + Send a request to the appropriate tool shed to retrieve the dictionary of repository dependencies defined + for the received repository which is installed into Galaxy. This method is called only from Galaxy. + """ + tool_shed_url = common_util.get_tool_shed_url_from_tool_shed_registry( app, str( repository.tool_shed ) ) + params = '?name=%s&owner=%s&changeset_revision=%s' % ( str( repository.name ), + str( repository.owner ), + str( repository.changeset_revision ) ) + url = common_util.url_join( tool_shed_url, + 'repository/get_repository_dependencies%s' % params ) + try: + raw_text = common_util.tool_shed_get( app, tool_shed_url, url ) + except Exception, e: + print "The URL\n%s\nraised the exception:\n%s\n" % ( url, str( e ) ) + return '' + if len( raw_text ) > 2: + encoded_text = json.loads( raw_text ) + text = encoding_util.tool_shed_decode( encoded_text ) + else: + text = '' + return text + def get_repository_dependency_by_repository_id( self, install_model, decoded_repository_id ): return install_model.context.query( install_model.RepositoryDependency ) \ .filter( install_model.RepositoryDependency.table.c.tool_shed_repository_id == decoded_repository_id ) \ diff -r 344e84a6554ae94604f32e8f62feebd65a6ac358 -r f5589d243d0f920cdd96c3e980762bb1a7cc061b lib/tool_shed/util/repository_dependency_util.py --- a/lib/tool_shed/util/repository_dependency_util.py +++ b/lib/tool_shed/util/repository_dependency_util.py @@ -147,29 +147,6 @@ prior_installation_required, only_if_compiling_contained_td ) -def get_repository_dependencies_for_installed_tool_shed_repository( app, repository ): - """ - Send a request to the appropriate tool shed to retrieve the dictionary of repository dependencies defined - for the received repository which is installed into Galaxy. This method is called only from Galaxy. - """ - tool_shed_url = common_util.get_tool_shed_url_from_tool_shed_registry( app, str( repository.tool_shed ) ) - params = '?name=%s&owner=%s&changeset_revision=%s' % ( str( repository.name ), - str( repository.owner ), - str( repository.changeset_revision ) ) - url = common_util.url_join( tool_shed_url, - 'repository/get_repository_dependencies%s' % params ) - try: - raw_text = common_util.tool_shed_get( app, tool_shed_url, url ) - except Exception, e: - print "The URL\n%s\nraised the exception:\n%s\n" % ( url, str( e ) ) - return '' - if len( raw_text ) > 2: - encoded_text = json.loads( raw_text ) - text = encoding_util.tool_shed_decode( encoded_text ) - else: - text = '' - return text - def get_repository_dependencies_for_changeset_revision( app, repository, repository_metadata, toolshed_base_url, key_rd_dicts_to_be_processed=None, all_repository_dependencies=None, handled_key_rd_dicts=None, circular_repository_dependencies=None ): Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: greg: Eliminate the use of trans when exporting repository capsules from the Tool SHed and move the new tag_attribute_haldler.py module where it was meant to be located.
by commits-noreply＠bitbucket.org 22 Jun '14

22 Jun '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/344e84a6554a/ Changeset: 344e84a6554a User: greg Date: 2014-06-23 00:25:22 Summary: Eliminate the use of trans when exporting repository capsules from the Tool SHed and move the new tag_attribute_haldler.py module where it was meant to be located. Affected #: 7 files diff -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 -r 344e84a6554ae94604f32e8f62feebd65a6ac358 lib/galaxy/webapps/tool_shed/api/repository_revisions.py --- a/lib/galaxy/webapps/tool_shed/api/repository_revisions.py +++ b/lib/galaxy/webapps/tool_shed/api/repository_revisions.py @@ -53,7 +53,8 @@ log.debug( error_message ) return None, error_message repository_id = trans.security.encode_id( repository.id ) - return export_util.export_repository( trans, + return export_util.export_repository( trans.app, + trans.user, tool_shed_url, repository_id, str( repository.name ), diff -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 -r 344e84a6554ae94604f32e8f62feebd65a6ac358 lib/galaxy/webapps/tool_shed/controllers/repository.py --- a/lib/galaxy/webapps/tool_shed/controllers/repository.py +++ b/lib/galaxy/webapps/tool_shed/controllers/repository.py @@ -1172,7 +1172,8 @@ file_type = 'gz' export_repository_dependencies = CheckboxField.is_checked( export_repository_dependencies ) tool_shed_url = web.url_for( '/', qualified=True ) - repositories_archive, error_message = export_util.export_repository( trans, + repositories_archive, error_message = export_util.export_repository( trans.app, + trans.user, tool_shed_url, repository_id, str( repository.name ), diff -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 -r 344e84a6554ae94604f32e8f62feebd65a6ac358 lib/tool_shed/dependencies/dependency_manager.py --- a/lib/tool_shed/dependencies/dependency_manager.py +++ b/lib/tool_shed/dependencies/dependency_manager.py @@ -5,7 +5,7 @@ from galaxy.util.odict import odict from galaxy.web import url_for -from tool_shed.dependencies import tag_attribute_handler +from tool_shed.dependencies.tool import tag_attribute_handler from tool_shed.repository_types.util import REPOSITORY_DEPENDENCY_DEFINITION_FILENAME from tool_shed.repository_types.util import TOOL_DEPENDENCY_DEFINITION_FILENAME from tool_shed.util import hg_util diff -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 -r 344e84a6554ae94604f32e8f62feebd65a6ac358 lib/tool_shed/dependencies/tag_attribute_handler.py --- a/lib/tool_shed/dependencies/tag_attribute_handler.py +++ /dev/null @@ -1,199 +0,0 @@ -import copy -import logging - -log = logging.getLogger( __name__ ) - - -class TagAttributeHandler( object ): - - def __init__( self, app, rdd, unpopulate ): - self.app = app - self.altered = False - self.rdd = rdd - self.unpopulate = unpopulate - - def process_action_tag_set( self, elem, message ): - # Here we're inside of an <actions> tag set. See http://localhost:9009/view/devteam/package_r_2_11_0 . - # <action> - # <repository name="package_readline_6_2" owner="devteam"> - # <package name="readline" version="6.2" /> - # </repository> - # </action> - elem_altered = False - new_elem = copy.deepcopy( elem ) - for sub_index, sub_elem in enumerate( elem ): - altered = False - error_message = '' - if sub_elem.tag == 'repository': - altered, new_sub_elem, error_message = \ - self.process_repository_tag_set( parent_elem=elem, - elem_index=sub_index, - elem=sub_elem, - message=message ) - if error_message: - message += error_message - if altered: - if not self.altered: - self.altered = True - if not elem_altered: - elem_altered = True - new_elem[ sub_index ] = new_sub_elem - return elem_altered, new_elem, message - - def process_actions_tag_set( self, elem, message ): - # <actions> - # <package name="libgtextutils" version="0.6"> - # <repository name="package_libgtextutils_0_6" owner="test" prior_installation_required="True" /> - # </package> - from tool_shed.util import xml_util - elem_altered = False - new_elem = copy.deepcopy( elem ) - for sub_index, sub_elem in enumerate( elem ): - altered = False - error_message = '' - if sub_elem.tag == 'package': - altered, new_sub_elem, error_message = self.process_package_tag_set( elem=sub_elem, - message=message ) - elif sub_elem.tag == 'action': - # <action type="set_environment_for_install"> - # <repository name="package_readline_6_2" owner="devteam""> - # <package name="readline" version="6.2" /> - # </repository> - # </action> - altered, new_sub_elem, error_message = self.process_action_tag_set( elem=sub_elem, - message=message ) - else: - # Inspect the sub elements of elem to locate all <repository> tags and - # populate them with toolshed and changeset_revision attributes if necessary. - altered, new_sub_elem, error_message = self.rdd.handle_sub_elem( parent_elem=elem, - elem_index=sub_index, - elem=sub_elem ) - if error_message: - message += error_message - if altered: - if not self.altered: - self.altered = True - if not elem_altered: - elem_altered = True - new_elem[ sub_index ] = new_sub_elem - return elem_altered, new_elem, message - - def process_actions_group_tag_set( self, elem, message, skip_actions_tags=False ): - # Inspect all entries in the <actions_group> tag set, skipping <actions> - # tag sets that define os and architecture attributes. We want to inspect - # only the last <actions> tag set contained within the <actions_group> tag - # set to see if a complex repository dependency is defined. - elem_altered = False - new_elem = copy.deepcopy( elem ) - for sub_index, sub_elem in enumerate( elem ): - altered = False - error_message = '' - if sub_elem.tag == 'actions': - if skip_actions_tags: - # Skip all actions tags that include os or architecture attributes. - system = sub_elem.get( 'os' ) - architecture = sub_elem.get( 'architecture' ) - if system or architecture: - continue - altered, new_sub_elem, error_message = \ - self.process_actions_tag_set( elem=sub_elem, - message=message ) - if error_message: - message += error_message - if altered: - if not self.altered: - self.altered = True - if not elem_altered: - elem_altered = True - new_elem[ sub_index ] = new_sub_elem - return elem_altered, new_elem, message - - def process_config( self, root ): - error_message = '' - new_root = copy.deepcopy( root ) - if root.tag == 'tool_dependency': - for elem_index, elem in enumerate( root ): - altered = False - if elem.tag == 'package': - # <package name="eigen" version="2.0.17"> - altered, new_elem, error_message = \ - self.process_package_tag_set( elem=elem, - message=error_message ) - if altered: - if not self.altered: - self.altered = True - new_root[ elem_index ] = new_elem - else: - error_message = "Invalid tool_dependencies.xml file." - return self.altered, new_root, error_message - - def process_install_tag_set( self, elem, message ): - # <install version="1.0"> - elem_altered = False - new_elem = copy.deepcopy( elem ) - for sub_index, sub_elem in enumerate( elem ): - altered = False - error_message = '' - if sub_elem.tag == 'actions_group': - altered, new_sub_elem, error_message = \ - self.process_actions_group_tag_set( elem=sub_elem, - message=message, - skip_actions_tags=True ) - elif sub_elem.tag == 'actions': - altered, new_sub_elem, error_message = \ - self.process_actions_tag_set( elem=sub_elem, - message=message ) - else: - package_name = elem.get( 'name', '' ) - package_version = elem.get( 'version', '' ) - error_message += 'Version %s of the %s package cannot be installed because ' % \ - ( str( package_version ), str( package_name ) ) - error_message += 'the recipe for installing the package is missing either an ' - error_message += '<actions> tag set or an <actions_group> tag set.' - if error_message: - message += error_message - if altered: - if not self.altered: - self.altered = True - if not elem_altered: - elem_altered = True - new_elem[ sub_index ] = new_sub_elem - return elem_altered, new_elem, message - - def process_package_tag_set( self, elem, message ): - elem_altered = False - new_elem = copy.deepcopy( elem ) - for sub_index, sub_elem in enumerate( elem ): - altered = False - error_message = '' - if sub_elem.tag == 'install': - altered, new_sub_elem, error_message = \ - self.process_install_tag_set( elem=sub_elem, - message=message ) - elif sub_elem.tag == 'repository': - altered, new_sub_elem, error_message = \ - self.process_repository_tag_set( parent_elem=elem, - elem_index=sub_index, - elem=sub_elem, - message=message ) - if error_message: - message += error_message - if altered: - if not self.altered: - self.altered = True - if not elem_altered: - elem_altered = True - new_elem[ sub_index ] = new_sub_elem - return elem_altered, new_elem, message - - def process_repository_tag_set( self, parent_elem, elem_index, elem, message ): - # We have a complex repository dependency. - altered, new_elem, error_message = self.rdd.handle_complex_dependency_elem( parent_elem=parent_elem, - elem_index=elem_index, - elem=elem ) - if error_message: - message += error_message - if altered: - if not self.altered: - self.altered = True - return altered, new_elem, message diff -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 -r 344e84a6554ae94604f32e8f62feebd65a6ac358 lib/tool_shed/dependencies/tool/tag_attribute_handler.py --- /dev/null +++ b/lib/tool_shed/dependencies/tool/tag_attribute_handler.py @@ -0,0 +1,199 @@ +import copy +import logging + +log = logging.getLogger( __name__ ) + + +class TagAttributeHandler( object ): + + def __init__( self, app, rdd, unpopulate ): + self.app = app + self.altered = False + self.rdd = rdd + self.unpopulate = unpopulate + + def process_action_tag_set( self, elem, message ): + # Here we're inside of an <actions> tag set. See http://localhost:9009/view/devteam/package_r_2_11_0 . + # <action> + # <repository name="package_readline_6_2" owner="devteam"> + # <package name="readline" version="6.2" /> + # </repository> + # </action> + elem_altered = False + new_elem = copy.deepcopy( elem ) + for sub_index, sub_elem in enumerate( elem ): + altered = False + error_message = '' + if sub_elem.tag == 'repository': + altered, new_sub_elem, error_message = \ + self.process_repository_tag_set( parent_elem=elem, + elem_index=sub_index, + elem=sub_elem, + message=message ) + if error_message: + message += error_message + if altered: + if not self.altered: + self.altered = True + if not elem_altered: + elem_altered = True + new_elem[ sub_index ] = new_sub_elem + return elem_altered, new_elem, message + + def process_actions_tag_set( self, elem, message ): + # <actions> + # <package name="libgtextutils" version="0.6"> + # <repository name="package_libgtextutils_0_6" owner="test" prior_installation_required="True" /> + # </package> + from tool_shed.util import xml_util + elem_altered = False + new_elem = copy.deepcopy( elem ) + for sub_index, sub_elem in enumerate( elem ): + altered = False + error_message = '' + if sub_elem.tag == 'package': + altered, new_sub_elem, error_message = self.process_package_tag_set( elem=sub_elem, + message=message ) + elif sub_elem.tag == 'action': + # <action type="set_environment_for_install"> + # <repository name="package_readline_6_2" owner="devteam""> + # <package name="readline" version="6.2" /> + # </repository> + # </action> + altered, new_sub_elem, error_message = self.process_action_tag_set( elem=sub_elem, + message=message ) + else: + # Inspect the sub elements of elem to locate all <repository> tags and + # populate them with toolshed and changeset_revision attributes if necessary. + altered, new_sub_elem, error_message = self.rdd.handle_sub_elem( parent_elem=elem, + elem_index=sub_index, + elem=sub_elem ) + if error_message: + message += error_message + if altered: + if not self.altered: + self.altered = True + if not elem_altered: + elem_altered = True + new_elem[ sub_index ] = new_sub_elem + return elem_altered, new_elem, message + + def process_actions_group_tag_set( self, elem, message, skip_actions_tags=False ): + # Inspect all entries in the <actions_group> tag set, skipping <actions> + # tag sets that define os and architecture attributes. We want to inspect + # only the last <actions> tag set contained within the <actions_group> tag + # set to see if a complex repository dependency is defined. + elem_altered = False + new_elem = copy.deepcopy( elem ) + for sub_index, sub_elem in enumerate( elem ): + altered = False + error_message = '' + if sub_elem.tag == 'actions': + if skip_actions_tags: + # Skip all actions tags that include os or architecture attributes. + system = sub_elem.get( 'os' ) + architecture = sub_elem.get( 'architecture' ) + if system or architecture: + continue + altered, new_sub_elem, error_message = \ + self.process_actions_tag_set( elem=sub_elem, + message=message ) + if error_message: + message += error_message + if altered: + if not self.altered: + self.altered = True + if not elem_altered: + elem_altered = True + new_elem[ sub_index ] = new_sub_elem + return elem_altered, new_elem, message + + def process_config( self, root ): + error_message = '' + new_root = copy.deepcopy( root ) + if root.tag == 'tool_dependency': + for elem_index, elem in enumerate( root ): + altered = False + if elem.tag == 'package': + # <package name="eigen" version="2.0.17"> + altered, new_elem, error_message = \ + self.process_package_tag_set( elem=elem, + message=error_message ) + if altered: + if not self.altered: + self.altered = True + new_root[ elem_index ] = new_elem + else: + error_message = "Invalid tool_dependencies.xml file." + return self.altered, new_root, error_message + + def process_install_tag_set( self, elem, message ): + # <install version="1.0"> + elem_altered = False + new_elem = copy.deepcopy( elem ) + for sub_index, sub_elem in enumerate( elem ): + altered = False + error_message = '' + if sub_elem.tag == 'actions_group': + altered, new_sub_elem, error_message = \ + self.process_actions_group_tag_set( elem=sub_elem, + message=message, + skip_actions_tags=True ) + elif sub_elem.tag == 'actions': + altered, new_sub_elem, error_message = \ + self.process_actions_tag_set( elem=sub_elem, + message=message ) + else: + package_name = elem.get( 'name', '' ) + package_version = elem.get( 'version', '' ) + error_message += 'Version %s of the %s package cannot be installed because ' % \ + ( str( package_version ), str( package_name ) ) + error_message += 'the recipe for installing the package is missing either an ' + error_message += '<actions> tag set or an <actions_group> tag set.' + if error_message: + message += error_message + if altered: + if not self.altered: + self.altered = True + if not elem_altered: + elem_altered = True + new_elem[ sub_index ] = new_sub_elem + return elem_altered, new_elem, message + + def process_package_tag_set( self, elem, message ): + elem_altered = False + new_elem = copy.deepcopy( elem ) + for sub_index, sub_elem in enumerate( elem ): + altered = False + error_message = '' + if sub_elem.tag == 'install': + altered, new_sub_elem, error_message = \ + self.process_install_tag_set( elem=sub_elem, + message=message ) + elif sub_elem.tag == 'repository': + altered, new_sub_elem, error_message = \ + self.process_repository_tag_set( parent_elem=elem, + elem_index=sub_index, + elem=sub_elem, + message=message ) + if error_message: + message += error_message + if altered: + if not self.altered: + self.altered = True + if not elem_altered: + elem_altered = True + new_elem[ sub_index ] = new_sub_elem + return elem_altered, new_elem, message + + def process_repository_tag_set( self, parent_elem, elem_index, elem, message ): + # We have a complex repository dependency. + altered, new_elem, error_message = self.rdd.handle_complex_dependency_elem( parent_elem=parent_elem, + elem_index=elem_index, + elem=elem ) + if error_message: + message += error_message + if altered: + if not self.altered: + self.altered = True + return altered, new_elem, message diff -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 -r 344e84a6554ae94604f32e8f62feebd65a6ac358 lib/tool_shed/util/export_util.py --- a/lib/tool_shed/util/export_util.py +++ b/lib/tool_shed/util/export_util.py @@ -7,7 +7,6 @@ from time import gmtime from time import strftime import tool_shed.repository_types.util as rt_util -from galaxy import eggs from galaxy import web from galaxy.util.odict import odict from tool_shed.dependencies import dependency_manager @@ -21,11 +20,6 @@ from tool_shed.util import xml_util from tool_shed.galaxy_install.repository_dependencies.repository_dependency_manager import RepositoryDependencyManager -eggs.require( 'mercurial' ) - -from mercurial import commands -from mercurial import ui - log = logging.getLogger( __name__ ) CAPSULE_FILENAME = 'capsule' @@ -36,24 +30,9 @@ def __init__( self ): self.exported_repository_elems = [] -def archive_repository_revision( trans, ui, repository, archive_dir, changeset_revision ): - '''Create an un-versioned archive of a repository.''' - repo = hg_util.get_repo_for_repository( trans.app, repository=repository, repo_path=None, create=False ) - options_dict = hg_util.get_mercurial_default_options_dict( 'archive' ) - options_dict[ 'rev' ] = changeset_revision - error_message = '' - return_code = None - try: - return_code = commands.archive( ui, repo, archive_dir, **options_dict ) - except Exception, e: - error_message = "Error attempting to archive revision <b>%s</b> of repository %s: %s\nReturn code: %s\n" % \ - ( str( changeset_revision ), str( repository.name ), str( e ), str( return_code ) ) - log.exception( error_message ) - return return_code, error_message - -def export_repository( trans, tool_shed_url, repository_id, repository_name, changeset_revision, file_type, +def export_repository( app, user, tool_shed_url, repository_id, repository_name, changeset_revision, file_type, export_repository_dependencies, api=False ): - repository = suc.get_repository_in_tool_shed( trans.app, repository_id ) + repository = suc.get_repository_in_tool_shed( app, repository_id ) repositories_archive_filename = generate_repository_archive_filename( tool_shed_url, str( repository.name ), str( repository.user.username ), @@ -62,16 +41,16 @@ export_repository_dependencies=export_repository_dependencies, use_tmp_archive_dir=True ) if export_repository_dependencies: - repo_info_dicts = get_repo_info_dicts( trans, tool_shed_url, repository_id, changeset_revision ) - repository_ids = get_repository_ids( trans, repo_info_dicts ) + repo_info_dicts = get_repo_info_dicts( app, user, tool_shed_url, repository_id, changeset_revision ) + repository_ids = get_repository_ids( app, repo_info_dicts ) ordered_repository_ids, ordered_repositories, ordered_changeset_revisions = \ - order_components_for_import( trans, repository_id, repository_ids, repo_info_dicts ) + order_components_for_import( app, repository_id, repository_ids, repo_info_dicts ) else: ordered_repository_ids = [] ordered_repositories = [] ordered_changeset_revisions = [] if repository: - repository_metadata = suc.get_current_repository_metadata_for_changeset_revision( trans.app, + repository_metadata = suc.get_current_repository_metadata_for_changeset_revision( app, repository, changeset_revision ) if repository_metadata: @@ -89,7 +68,7 @@ work_dir = tempfile.mkdtemp( prefix="tmp-toolshed-export-er" ) ordered_repository = ordered_repositories[ index ] ordered_changeset_revision = ordered_changeset_revisions[ index ] - repository_archive, error_message = generate_repository_archive( trans, + repository_archive, error_message = generate_repository_archive( app, work_dir, tool_shed_url, ordered_repository, @@ -128,12 +107,15 @@ return dict( download_url=download_url, error_messages=error_messages ) return repositories_archive, error_messages -def generate_repository_archive( trans, work_dir, tool_shed_url, repository, changeset_revision, file_type ): - rdah = dependency_manager.RepositoryDependencyAttributeHandler( trans.app, unpopulate=True ) - tdah = dependency_manager.ToolDependencyAttributeHandler( trans.app, unpopulate=True ) +def generate_repository_archive( app, work_dir, tool_shed_url, repository, changeset_revision, file_type ): + rdah = dependency_manager.RepositoryDependencyAttributeHandler( app, unpopulate=True ) + tdah = dependency_manager.ToolDependencyAttributeHandler( app, unpopulate=True ) file_type_str = get_file_type_str( changeset_revision, file_type ) file_name = '%s-%s' % ( repository.name, file_type_str ) - return_code, error_message = archive_repository_revision( trans, ui, repository, work_dir, changeset_revision ) + return_code, error_message = hg_util.archive_repository_revision( app, + repository, + work_dir, + changeset_revision ) if return_code: return None, error_message repository_archive_name = os.path.join( work_dir, file_name ) @@ -195,7 +177,7 @@ sub_elements[ 'exported_via_api' ] = str( api ) return sub_elements -def get_components_from_repo_info_dict( trans, repo_info_dict ): +def get_components_from_repo_info_dict( app, repo_info_dict ): """ Return the repository and the associated latest installable changeset_revision (including updates) for the repository defined by the received repo_info_dict. @@ -204,8 +186,8 @@ # There should only be one entry in the received repo_info_dict. description, repository_clone_url, changeset_revision, ctx_rev, repository_owner, repository_dependencies, tool_dependencies = \ suc.get_repo_info_tuple_contents( repo_info_tup ) - repository = suc.get_repository_by_name_and_owner( trans.app, repository_name, repository_owner ) - repository_metadata = suc.get_current_repository_metadata_for_changeset_revision( trans.app, + repository = suc.get_repository_by_name_and_owner( app, repository_name, repository_owner ) + repository_metadata = suc.get_current_repository_metadata_for_changeset_revision( app, repository, changeset_revision ) if repository_metadata: @@ -235,30 +217,30 @@ return repo_info_dict return None -def get_repo_info_dicts( trans, tool_shed_url, repository_id, changeset_revision ): +def get_repo_info_dicts( app, user, tool_shed_url, repository_id, changeset_revision ): """ Return a list of dictionaries defining repositories that are required by the repository associated with the received repository_id. """ - rdm = RepositoryDependencyManager( trans.app ) - repository = suc.get_repository_in_tool_shed( trans.app, repository_id ) - repository_metadata = suc.get_repository_metadata_by_changeset_revision( trans.app, repository_id, changeset_revision ) + rdm = RepositoryDependencyManager( app ) + repository = suc.get_repository_in_tool_shed( app, repository_id ) + repository_metadata = suc.get_repository_metadata_by_changeset_revision( app, repository_id, changeset_revision ) # Get a dictionary of all repositories upon which the contents of the current repository_metadata record depend. toolshed_base_url = str( web.url_for( '/', qualified=True ) ).rstrip( '/' ) repository_dependencies = \ - repository_dependency_util.get_repository_dependencies_for_changeset_revision( app=trans.app, + repository_dependency_util.get_repository_dependencies_for_changeset_revision( app=app, repository=repository, repository_metadata=repository_metadata, toolshed_base_url=toolshed_base_url, key_rd_dicts_to_be_processed=None, all_repository_dependencies=None, handled_key_rd_dicts=None ) - repo = hg_util.get_repo_for_repository( trans.app, repository=repository, repo_path=None, create=False ) + repo = hg_util.get_repo_for_repository( app, repository=repository, repo_path=None, create=False ) ctx = hg_util.get_changectx_for_changeset( repo, changeset_revision ) repo_info_dict = {} # Cast unicode to string. repo_info_dict[ str( repository.name ) ] = ( str( repository.description ), - common_util.generate_clone_url_for_repository_in_tool_shed( trans.user, repository ), + common_util.generate_clone_url_for_repository_in_tool_shed( user, repository ), str( changeset_revision ), str( ctx.rev() ), str( repository.user.username ), @@ -292,18 +274,24 @@ sub_elements[ 'categories' ] = categories return attributes, sub_elements -def get_repository_ids( trans, repo_info_dicts ): +def get_repository_ids( app, repo_info_dicts ): """Return a list of repository ids associated with each dictionary in the received repo_info_dicts.""" repository_ids = [] for repo_info_dict in repo_info_dicts: for repository_name, repo_info_tup in repo_info_dict.items(): - description, repository_clone_url, changeset_revision, ctx_rev, repository_owner, repository_dependencies, tool_dependencies = \ - suc.get_repo_info_tuple_contents( repo_info_tup ) - repository = suc.get_repository_by_name_and_owner( trans.app, repository_name, repository_owner ) - repository_ids.append( trans.security.encode_id( repository.id ) ) + description, \ + repository_clone_url, \ + changeset_revision, \ + ctx_rev, \ + repository_owner, \ + repository_dependencies, \ + tool_dependencies = \ + suc.get_repo_info_tuple_contents( repo_info_tup ) + repository = suc.get_repository_by_name_and_owner( app, repository_name, repository_owner ) + repository_ids.append( app.security.encode_id( repository.id ) ) return repository_ids -def order_components_for_import( trans, primary_repository_id, repository_ids, repo_info_dicts ): +def order_components_for_import( app, primary_repository_id, repository_ids, repo_info_dicts ): """ Some repositories may have repository dependencies that must be imported and have metadata set on them before the dependent repository is imported. This method will inspect the list of repositories @@ -322,7 +310,7 @@ # Create a dictionary whose keys are the received repository_ids and whose values are a list of # repository_ids, each of which is contained in the received list of repository_ids and whose associated # repository must be imported prior to the repository associated with the repository_id key. - prior_import_required_dict = suc.get_prior_import_or_install_required_dict( trans.app, repository_ids, repo_info_dicts ) + prior_import_required_dict = suc.get_prior_import_or_install_required_dict( app, repository_ids, repo_info_dicts ) processed_repository_ids = [] # Process the list of repository dependencies defined for the repository associated with the received # primary_repository_id. @@ -340,20 +328,20 @@ if prior_import_required_id not in ordered_repository_ids: # Import the associated repository dependency first. prior_repo_info_dict = get_repo_info_dict_for_import( prior_import_required_id, repository_ids, repo_info_dicts ) - prior_repository, prior_import_changeset_revision = get_components_from_repo_info_dict( trans, prior_repo_info_dict ) + prior_repository, prior_import_changeset_revision = get_components_from_repo_info_dict( app, prior_repo_info_dict ) if prior_repository and prior_import_changeset_revision: ordered_repository_ids.append( prior_import_required_id ) ordered_repositories.append( prior_repository ) ordered_changeset_revisions.append( prior_import_changeset_revision ) repo_info_dict = get_repo_info_dict_for_import( repository_id, repository_ids, repo_info_dicts ) - repository, changeset_revision = get_components_from_repo_info_dict( trans, repo_info_dict ) + repository, changeset_revision = get_components_from_repo_info_dict( app, repo_info_dict ) if repository and changeset_revision: ordered_repository_ids.append( repository_id ) ordered_repositories.append( repository ) ordered_changeset_revisions.append( changeset_revision ) # Process the repository associated with the received primary_repository_id last. repo_info_dict = get_repo_info_dict_for_import( primary_repository_id, repository_ids, repo_info_dicts ) - repository, changeset_revision = get_components_from_repo_info_dict( trans, repo_info_dict ) + repository, changeset_revision = get_components_from_repo_info_dict( app, repo_info_dict ) if repository and changeset_revision: ordered_repository_ids.append( repository_id ) ordered_repositories.append( repository ) diff -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 -r 344e84a6554ae94604f32e8f62feebd65a6ac358 lib/tool_shed/util/hg_util.py --- a/lib/tool_shed/util/hg_util.py +++ b/lib/tool_shed/util/hg_util.py @@ -28,6 +28,21 @@ def add_changeset( repo_ui, repo, path_to_filename_in_archive ): commands.add( repo_ui, repo, path_to_filename_in_archive ) +def archive_repository_revision( app, repository, archive_dir, changeset_revision ): + '''Create an un-versioned archive of a repository.''' + repo = get_repo_for_repository( app, repository=repository, repo_path=None, create=False ) + options_dict = get_mercurial_default_options_dict( 'archive' ) + options_dict[ 'rev' ] = changeset_revision + error_message = '' + return_code = None + try: + return_code = commands.archive( get_configured_ui, repo, archive_dir, **options_dict ) + except Exception, e: + error_message = "Error attempting to archive revision <b>%s</b> of repository %s: %s\nReturn code: %s\n" % \ + ( str( changeset_revision ), str( repository.name ), str( e ), str( return_code ) ) + log.exception( error_message ) + return return_code, error_message + def bundle_to_json( fh ): """ Convert the received HG10xx data stream (a mercurial 1.0 bundle created using hg push from the Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: greg: Move 2 Tool Shed utility functions into the middleware hg class.
by commits-noreply＠bitbucket.org 22 Jun '14

22 Jun '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/ae53980fad82/ Changeset: ae53980fad82 User: greg Date: 2014-06-22 14:00:57 Summary: Move 2 Tool Shed utility functions into the middleware hg class. Affected #: 3 files diff -r 63919647ca8424a28042ccb79ea2db6e0c68d507 -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 lib/galaxy/webapps/tool_shed/framework/middleware/hg.py --- a/lib/galaxy/webapps/tool_shed/framework/middleware/hg.py +++ b/lib/galaxy/webapps/tool_shed/framework/middleware/hg.py @@ -11,7 +11,6 @@ from galaxy.util import asbool from galaxy.util.hash_util import new_secure_hash -from tool_shed.util import commit_util from tool_shed.util import hg_util import tool_shed.repository_types.util as rt_util @@ -54,13 +53,15 @@ connection = engine.connect() path_info = environ[ 'PATH_INFO' ].lstrip( '/' ) user_id, repository_name = self.__get_user_id_repository_name_from_path_info( connection, path_info ) - sql_cmd = "SELECT times_downloaded FROM repository WHERE user_id = %d AND name = '%s'" % ( user_id, repository_name.lower() ) + sql_cmd = "SELECT times_downloaded FROM repository WHERE user_id = %d AND name = '%s'" % \ + ( user_id, repository_name.lower() ) result_set = connection.execute( sql_cmd ) for row in result_set: # Should only be 1 row... times_downloaded = row[ 'times_downloaded' ] times_downloaded += 1 - sql_cmd = "UPDATE repository SET times_downloaded = %d WHERE user_id = %d AND name = '%s'" % ( times_downloaded, user_id, repository_name.lower() ) + sql_cmd = "UPDATE repository SET times_downloaded = %d WHERE user_id = %d AND name = '%s'" % \ + ( times_downloaded, user_id, repository_name.lower() ) connection.execute( sql_cmd ) connection.close() elif cmd in [ 'unbundle', 'pushkey' ]: @@ -132,7 +133,7 @@ if filename and isinstance( filename, str ): if filename == rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME: # Make sure the any complex repository dependency definitions contain valid <repository> tags. - is_valid, error_msg = commit_util.repository_tags_are_valid( filename, change_list ) + is_valid, error_msg = self.repository_tags_are_valid( filename, change_list ) if not is_valid: log.debug( error_msg ) return self.__display_exception_remotely( start_response, error_msg ) @@ -151,7 +152,7 @@ if filename and isinstance( filename, str ): if filename == rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME: # Make sure the any complex repository dependency definitions contain valid <repository> tags. - is_valid, error_msg = commit_util.repository_tags_are_valid( filename, change_list ) + is_valid, error_msg = self.repository_tags_are_valid( filename, change_list ) if not is_valid: log.debug( error_msg ) return self.__display_exception_remotely( start_response, error_msg ) @@ -174,7 +175,7 @@ rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME ]: # We check both files since tool dependency definitions files can contain complex # repository dependency definitions. - is_valid, error_msg = commit_util.repository_tags_are_valid( filename, change_list ) + is_valid, error_msg = self.repository_tags_are_valid( filename, change_list ) if not is_valid: log.debug( error_msg ) return self.__display_exception_remotely( start_response, error_msg ) @@ -186,6 +187,54 @@ return result.wsgi_application( environ, start_response ) return self.app( environ, start_response ) + def __authenticate( self, username, password ): + db_password = None + # Instantiate a database connection + engine = sqlalchemy.create_engine( self.db_url ) + connection = engine.connect() + result_set = connection.execute( "select email, password from galaxy_user where username = '%s'" % username.lower() ) + for row in result_set: + # Should only be 1 row... + db_email = row[ 'email' ] + db_password = row[ 'password' ] + connection.close() + if db_password: + # Check if password matches db_password when hashed. + return new_secure_hash( text_type=password ) == db_password + return False + + def __authenticate_remote_user( self, environ, username, password ): + """ + Look after a remote user and "authenticate" - upstream server should already have achieved + this for us, but we check that the user exists at least. Hg allow_push = must include username + - some versions of mercurial blow up with 500 errors. + """ + db_username = None + ru_email = environ[ 'HTTP_REMOTE_USER' ].lower() + ## Instantiate a database connection... + engine = sqlalchemy.create_engine( self.db_url ) + connection = engine.connect() + result_set = connection.execute( "select email, username, password from galaxy_user where email = '%s'" % ru_email ) + for row in result_set: + # Should only be 1 row... + db_email = row[ 'email' ] + db_password = row[ 'password' ] + db_username = row[ 'username' ] + connection.close() + if db_username: + # We could check the password here except that the function galaxy.web.framework.get_or_create_remote_user() + # does some random generation of a password - so that no-one knows the password and only the hash is stored... + return db_username == username + return False + + def __basic_authentication( self, environ, username, password ): + """The environ parameter is needed in basic authentication. We also check it if use_remote_user is true.""" + if asbool( self.config.get( 'use_remote_user', False ) ): + assert "HTTP_REMOTE_USER" in environ, "use_remote_user is set but no HTTP_REMOTE_USER variable" + return self.__authenticate_remote_user( environ, username, password ) + else: + return self.__authenticate( username, password ) + def __display_exception_remotely( self, start_response, msg ): # Display the exception to the remote user's command line. status = "500 %s" % msg @@ -213,49 +262,37 @@ user_id = row[ 'id' ] return user_id, repository_name - def __basic_authentication( self, environ, username, password ): - """The environ parameter is needed in basic authentication. We also check it if use_remote_user is true.""" - if asbool( self.config.get( 'use_remote_user', False ) ): - assert "HTTP_REMOTE_USER" in environ, "use_remote_user is set but no HTTP_REMOTE_USER variable" - return self.__authenticate_remote_user( environ, username, password ) - else: - return self.__authenticate( username, password ) - - def __authenticate( self, username, password ): - db_password = None - # Instantiate a database connection - engine = sqlalchemy.create_engine( self.db_url ) - connection = engine.connect() - result_set = connection.execute( "select email, password from galaxy_user where username = '%s'" % username.lower() ) - for row in result_set: - # Should only be 1 row... - db_email = row[ 'email' ] - db_password = row[ 'password' ] - connection.close() - if db_password: - # Check if password matches db_password when hashed. - return new_secure_hash( text_type=password ) == db_password - return False - - def __authenticate_remote_user( self, environ, username, password ): + def repository_tag_is_valid( self, filename, line ): """ - Look after a remote user and "authenticate" - upstream server should already have achieved this for us, but we check that the - user exists at least. Hg allow_push = must include username - some versions of mercurial blow up with 500 errors. + Checks changes made to <repository> tags in a dependency definition file being pushed to the + Tool Shed from the command line to ensure that all required attributes exist. """ - db_username = None - ru_email = environ[ 'HTTP_REMOTE_USER' ].lower() - ## Instantiate a database connection... - engine = sqlalchemy.create_engine( self.db_url ) - connection = engine.connect() - result_set = connection.execute( "select email, username, password from galaxy_user where email = '%s'" % ru_email ) - for row in result_set: - # Should only be 1 row... - db_email = row[ 'email' ] - db_password = row[ 'password' ] - db_username = row[ 'username' ] - connection.close() - if db_username: - # We could check the password here except that the function galaxy.web.framework.get_or_create_remote_user() does some random generation of - # a password - so that no-one knows the password and only the hash is stored... - return db_username == username - return False + required_attributes = [ 'toolshed', 'name', 'owner', 'changeset_revision' ] + defined_attributes = line.split() + for required_attribute in required_attributes: + defined = False + for defined_attribute in defined_attributes: + if defined_attribute.startswith( required_attribute ): + defined = True + break + if not defined: + error_msg = 'The %s file contains a <repository> tag that is missing the required attribute %s. ' % \ + ( filename, required_attribute ) + error_msg += 'Automatically populating dependency definition attributes occurs only when using ' + error_msg += 'the Tool Shed upload utility. ' + return False, error_msg + return True, '' + + def repository_tags_are_valid( self, filename, change_list ): + """ + Make sure the any complex repository dependency definitions contain valid <repository> tags when pushing + changes to the tool shed on the command line. + """ + tag = '<repository' + for change_dict in change_list: + lines = get_change_lines_in_file_for_tag( tag, change_dict ) + for line in lines: + is_valid, error_msg = repository_tag_is_valid( filename, line ) + if not is_valid: + return False, error_msg + return True, '' diff -r 63919647ca8424a28042ccb79ea2db6e0c68d507 -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 lib/tool_shed/dependencies/dependency_manager.py --- a/lib/tool_shed/dependencies/dependency_manager.py +++ b/lib/tool_shed/dependencies/dependency_manager.py @@ -177,6 +177,7 @@ new_root[ index ] = new_elem return root_altered, new_root, error_message + class ToolDependencyAttributeHandler( object ): def __init__( self, app, unpopulate ): diff -r 63919647ca8424a28042ccb79ea2db6e0c68d507 -r ae53980fad820db494c67ddb5ad0ad5f9deb9dd1 lib/tool_shed/util/commit_util.py --- a/lib/tool_shed/util/commit_util.py +++ b/lib/tool_shed/util/commit_util.py @@ -221,41 +221,6 @@ gzipped_file.close() shutil.move( uncompressed, uploaded_file_name ) -def repository_tag_is_valid( filename, line ): - """ - Checks changes made to <repository> tags in a dependency definition file being pushed to the - Tool Shed from the command line to ensure that all required attributes exist. - """ - required_attributes = [ 'toolshed', 'name', 'owner', 'changeset_revision' ] - defined_attributes = line.split() - for required_attribute in required_attributes: - defined = False - for defined_attribute in defined_attributes: - if defined_attribute.startswith( required_attribute ): - defined = True - break - if not defined: - error_msg = 'The %s file contains a <repository> tag that is missing the required attribute %s. ' % \ - ( filename, required_attribute ) - error_msg += 'Automatically populating dependency definition attributes occurs only when using ' - error_msg += 'the Tool Shed upload utility. ' - return False, error_msg - return True, '' - -def repository_tags_are_valid( filename, change_list ): - """ - Make sure the any complex repository dependency definitions contain valid <repository> tags when pushing - changes to the tool shed on the command line. - """ - tag = '<repository' - for change_dict in change_list: - lines = get_change_lines_in_file_for_tag( tag, change_dict ) - for line in lines: - is_valid, error_msg = repository_tag_is_valid( filename, line ) - if not is_valid: - return False, error_msg - return True, '' - def uncompress( repository, uploaded_file_name, uploaded_file_filename, isgzip=False, isbz2=False ): if isgzip: handle_gzip( repository, uploaded_file_name ) Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0