Merge remote-tracking branch 'upstream/release_19.05' into dev

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)

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:

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(),

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

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()