diff --git a/lib/galaxy/job_metrics/__init__.py b/lib/galaxy/job_metrics/__init__.py index d99f4e06693..83f3c026ca6 100644 --- a/lib/galaxy/job_metrics/__init__.py +++ b/lib/galaxy/job_metrics/__init__.py @@ -108,15 +108,15 @@ class JobInstrumenter(object): return "\n".join([c for c in commands if c]) def collect_properties(self, job_id, job_directory): - per_plugin_properites = {} + per_plugin_properties = {} for plugin in self.plugins: try: properties = plugin.job_properties(job_id, job_directory) if properties: - per_plugin_properites[plugin.plugin_type] = properties + per_plugin_properties[plugin.plugin_type] = properties except Exception: log.exception("Failed to collect job properties for plugin %s", plugin) - return per_plugin_properites + return per_plugin_properties def __plugins_from_source(self, plugins_source): return plugin_config.load_plugins(self.plugin_classes, plugins_source, self.extra_kwargs) diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index 8d8283ae2a8..d6c003eea1d 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1399,6 +1399,7 @@ class JobWrapper(HasResourceParameters): job_stderr=None, check_output_detected_state=None, remote_metadata_directory=None, + job_metrics_directory=None, ): """ Called to indicate that the associated command has been run. Updates @@ -1638,7 +1639,7 @@ class JobWrapper(HasResourceParameters): job.set_final_state(final_job_state) if not job.tasks: # If job was composed of tasks, don't attempt to recollect statisitcs - self._collect_metrics(job) + self._collect_metrics(job, job_metrics_directory) self.sa_session.flush() log.debug('job %d ended (finish() executed in %s)' % (self.job_id, finish_timer)) if job.state == job.states.ERROR: @@ -1725,11 +1726,12 @@ class JobWrapper(HasResourceParameters): except Exception as e: log.debug("Error in collect_associated_files: %s" % (e)) - def _collect_metrics(self, has_metrics): + def _collect_metrics(self, has_metrics, job_metrics_directory=None): job = has_metrics.get_job() - per_plugin_properties = self.app.job_metrics.collect_properties(job.destination_id, self.job_id, self.working_directory) + job_metrics_directory = job_metrics_directory or self.working_directory + per_plugin_properties = self.app.job_metrics.collect_properties(job.destination_id, self.job_id, job_metrics_directory) if per_plugin_properties: - log.info("Collecting metrics for %s %s" % (type(has_metrics).__name__, getattr(has_metrics, 'id', None))) + log.info("Collecting metrics for %s %s in %s" % (type(has_metrics).__name__, getattr(has_metrics, 'id', None), job_metrics_directory)) for plugin, properties in per_plugin_properties.items(): for metric_name, metric_value in properties.items(): if metric_value is not None: diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py index eed1c5605aa..eda05554b6f 100644 --- a/lib/galaxy/jobs/runners/pulsar.py +++ b/lib/galaxy/jobs/runners/pulsar.py @@ -517,11 +517,18 @@ class PulsarJobRunner(AsynchronousJobRunner): self._handle_metadata_externally(job_wrapper, resolve_requirements=True) # Finish the job try: + job_metrics_directory = os.path.join(job_wrapper.working_directory, "metadata") + # Following check is a hack for jobs started during 19.01 or earlier release + # and finishing with a 19.05 code base. Eliminate the hack in 19.09 or later + # along with hacks for legacy metadata compute strategy. + if not os.path.exists(job_metrics_directory) or not any(["__instrument" in f for f in os.listdir(job_metrics_directory)]): + job_metrics_directory = job_wrapper.working_directory job_wrapper.finish( stdout, stderr, exit_code, remote_metadata_directory=remote_metadata_directory, + job_metrics_directory=job_metrics_directory, ) except Exception: log.exception("Job wrapper finish method failed") @@ -616,9 +623,14 @@ class PulsarJobRunner(AsynchronousJobRunner): 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) + if self.app.config.metadata_strategy == "legacy": + # Drop this branch in 19.09. + metadata_directory = job_wrapper.working_directory + else: + metadata_directory = os.path.join(job_wrapper.working_directory, "metadata") client_outputs = ClientOutputs( working_directory=job_wrapper.tool_working_directory, - metadata_directory=job_wrapper.working_directory, + metadata_directory=metadata_directory, work_dir_outputs=work_dir_outputs, output_files=output_files, version_file=job_wrapper.get_version_string_path(), diff --git a/lib/galaxy/metadata/__init__.py b/lib/galaxy/metadata/__init__.py index e7351771522..b095abe60cb 100644 --- a/lib/galaxy/metadata/__init__.py +++ b/lib/galaxy/metadata/__init__.py @@ -13,7 +13,7 @@ from six.moves import cPickle import galaxy.model from galaxy.model.metadata import FileParameter, MetadataTempFile -from galaxy.util import in_directory +from galaxy.util import in_directory, safe_makedirs log = getLogger(__name__) @@ -73,7 +73,11 @@ class MetadataCollectionStrategy(object): normalized_remote_metadata_directory = remote_metadata_directory and os.path.normpath(remote_metadata_directory) normalized_path = os.path.normpath(path) if remote_metadata_directory and normalized_path.startswith(normalized_remote_metadata_directory): - return normalized_path.replace(normalized_remote_metadata_directory, working_directory, 1) + if self.portable: + target_directory = os.path.join(working_directory, "metadata") + else: + target_directory = working_directory + return normalized_path.replace(normalized_remote_metadata_directory, target_directory, 1) return path dataset.metadata.from_JSON_dict(metadata_output_path, path_rewriter=path_rewriter) @@ -88,6 +92,7 @@ class MetadataCollectionStrategy(object): class PortableDirectoryMetadataGenerator(MetadataCollectionStrategy): + portable = True def __init__(self, job_id): self.job_id = job_id @@ -104,7 +109,8 @@ class PortableDirectoryMetadataGenerator(MetadataCollectionStrategy): tmp_dir = _init_tmp_dir(tmp_dir) metadata_dir = os.path.join(tmp_dir, "metadata") - os.mkdir(metadata_dir) + # may already exist (i.e. metadata collection in the job handler) + safe_makedirs(metadata_dir) def job_relative_path(path): path_relative = os.path.relpath(path, tmp_dir) @@ -171,6 +177,7 @@ class JobExternalOutputMetadataWrapper(MetadataCollectionStrategy): DatasetInstance object which will use pickle (in the future this could be JSONified as well) """ + portable = False def __init__(self, job_id): self.job_id = job_id @@ -393,8 +400,5 @@ def _get_filename_override(output_fnames, file_name): def _init_tmp_dir(tmp_dir): assert tmp_dir is not None - - if not os.path.exists(tmp_dir): - os.makedirs(tmp_dir) - + safe_makedirs(tmp_dir) return tmp_dir diff --git a/lib/galaxy_ext/metadata/set_metadata.py b/lib/galaxy_ext/metadata/set_metadata.py index a3f7ec9b844..9121bd8a4e6 100644 --- a/lib/galaxy_ext/metadata/set_metadata.py +++ b/lib/galaxy_ext/metadata/set_metadata.py @@ -77,7 +77,8 @@ def set_metadata(): def set_metadata_portable(): import galaxy.model - galaxy.model.metadata.MetadataTempFile.tmp_dir = tool_job_working_directory = os.path.abspath(os.getcwd()) + tool_job_working_directory = os.path.abspath(os.getcwd()) + galaxy.model.metadata.MetadataTempFile.tmp_dir = os.path.join(tool_job_working_directory, "metadata") metadata_params_path = os.path.join("metadata", "params.json") try: @@ -191,7 +192,11 @@ def validate_and_load_datatypes_config(datatypes_config): galaxy_root = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, os.pardir)) if not os.path.exists(datatypes_config): - print("Metadata setting failed because registry.xml could not be found. You may retry setting metadata.") + # Hack for Pulsar on usegalaxy.org, drop ASAP. + datatypes_config = "configs/registry.xml" + + if not os.path.exists(datatypes_config): + print("Metadata setting failed because registry.xml [%s] could not be found. You may retry setting metadata." % datatypes_config) sys.exit(1) import galaxy.datatypes.registry datatypes_registry = galaxy.datatypes.registry.Registry()