Introduce the concept of a metadata directory...

... for metadata about ... data and jobs. So formal Galaxy metadata stuff, job instrument files, etc....

Handle transfer of metadata and working directory outputs correctly in Galaxy 16.04.

Author: jmchilton

pulsar/client/action_mapper.py

@@ -39,12 +39,17 @@
     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..
+    # Input work dir files - e.g.task-split input file
     WORKDIR="workdir",
+    # Input work dir files - e.g. metadata files, etc..
+    METADATA="metadata",
     # 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",
+    # Meta job and data files (e.g. Galaxy metadata generation files and
+    # metric instrumentation files)
+    OUTPUT_METADATA="output_metadata",
     # Other fixed tool parameter paths (likely coming from tool data, but not
     # nessecarily). Not sure this is the best name...
     UNSTRUCTURED="unstructured",
@@ -56,8 +61,10 @@
     path_type.CONFIG,
     path_type.TOOL,
     path_type.WORKDIR,
+    path_type.METADATA,
     path_type.OUTPUT,
     path_type.OUTPUT_WORKDIR,
+    path_type.OUTPUT_METADATA,
 ]
 ALL_PATH_TYPES = ACTION_DEFAULT_PATH_TYPES + [path_type.UNSTRUCTURED]
 
@@ -206,7 +213,7 @@ 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":
+        if type in ["workdir", "output_workdir", "output_metadata"] 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"

pulsar/client/client.py

@@ -204,8 +204,8 @@ def fetch_output(self, path, name, working_directory, action_type, output_type):
             used if targetting an older Pulsar server that didn't return statuses
             allowing this to be inferred.
         """
-        if output_type == 'output_workdir':
-            self._fetch_work_dir_output(name, working_directory, path, action_type=action_type)
+        if output_type in ['output_workdir', 'output_metadata']:
+            self._populate_output_path(name, path, action_type, output_type)
         elif output_type == 'output':
             self._fetch_output(path=path, name=name, action_type=action_type)
         else:
@@ -219,22 +219,14 @@ def _fetch_output(self, path, name=None, check_exists_remotely=False, action_typ
             # Extra files will send in the path.
             name = os.path.basename(path)
 
-        self.__populate_output_path(name, path, action_type)
+        self._populate_output_path(name, path, action_type, path_type.OUTPUT)
 
-    def _fetch_work_dir_output(self, name, working_directory, output_path, action_type='transfer'):
+    def _populate_output_path(self, name, output_path, action_type, path_type):
         ensure_directory(output_path)
         if action_type == 'transfer':
-            self.__raw_download_output(name, self.job_id, path_type.OUTPUT_WORKDIR, output_path)
-        else:  # Even if action is none - Pulsar has a different work_dir so this needs to be copied.
-            pulsar_path = self._output_path(name, self.job_id, path_type.OUTPUT_WORKDIR)['path']
-            copy(pulsar_path, output_path)
-
-    def __populate_output_path(self, name, output_path, action_type):
-        ensure_directory(output_path)
-        if action_type == 'transfer':
-            self.__raw_download_output(name, self.job_id, path_type.OUTPUT, output_path)
+            self.__raw_download_output(name, self.job_id, path_type, output_path)
         elif action_type == 'copy':
-            pulsar_path = self._output_path(name, self.job_id, path_type.OUTPUT)['path']
+            pulsar_path = self._output_path(name, self.job_id, path_type)['path']
             copy(pulsar_path, output_path)
 
     @parseJson()

pulsar/client/job_directory.py

@@ -17,8 +17,10 @@
     config="configs_directory",
     tool="tool_files_directory",
     workdir="working_directory",
+    metadata="metadata_directory",
     output="outputs_directory",
     output_workdir="working_directory",
+    output_metadata="metadata_directory",
 )
 
 
@@ -33,6 +35,9 @@ def __init__(self, remote_staging_directory, remote_id, remote_sep):
             remote_id
         )
 
+    def metadata_directory(self):
+        return self._sub_dir('metadata')
+
     def working_directory(self):
         return self._sub_dir('working')
 
@@ -72,7 +77,7 @@ def _directory_for_file_type(self, file_type):
         # Obviously this client won't be legacy because this is in the
         # client module, but this code is reused on server which may
         # serve legacy clients.
-        allow_nested_files = file_type in ['input', 'unstructured', 'output', 'output_workdir']
+        allow_nested_files = file_type in ['input', 'unstructured', 'output', 'output_workdir', 'metadata', 'output_metadata']
         directory_function = getattr(self, TYPES_TO_METHOD.get(file_type, None), None)
         if not directory_function:
             raise Exception("Unknown file_type specified %s" % file_type)

pulsar/client/path_mapper.py

@@ -32,6 +32,7 @@ def __init__(
         self.input_directory = remote_job_config["inputs_directory"]
         self.output_directory = remote_job_config["outputs_directory"]
         self.working_directory = remote_job_config["working_directory"]
+        self.metadata_directory = remote_job_config.get("working_directory", None)
         self.unstructured_files_directory = remote_job_config["unstructured_files_directory"]
         self.config_directory = remote_job_config["configs_directory"]
         separator = remote_job_config["system_properties"]["separator"]
@@ -87,6 +88,8 @@ def __action(self, dataset_path, dataset_path_type):
     def __remote_directory(self, dataset_path_type):
         if dataset_path_type in [path_type.OUTPUT]:
             return self.output_directory
+        elif dataset_path_type in [path_type.METADATA, path_type.OUTPUT_METADATA]:
+            return self.metadata_directory
         elif dataset_path_type in [path_type.WORKDIR, path_type.OUTPUT_WORKDIR]:
             return self.working_directory
         elif dataset_path_type in [path_type.INPUT]:

pulsar/client/staging/__init__.py

@@ -31,7 +31,9 @@ class ClientJobDescription(object):
         Directory containing tool to execute (if a wrapper is used, it will
         be transferred to remote server).
     working_directory : str
-        Local path created by Galaxy for running this job.
+        Local path created by Galaxy for running this job (job_wrapper.tool_working_directory).
+    metadata_directory : str
+        Local path created by Galaxy for running this job (job_wrapper.working_directory).
     dependencies_description : list
         galaxy.tools.deps.dependencies.DependencyDescription object describing
         tool dependency context for remote depenency resolution.
@@ -57,7 +59,8 @@ def __init__(
         config_files=[],
         input_files=[],
         client_outputs=None,
-        working_directory=None,  # More sensible default?
+        working_directory=None,
+        metadata_directory=None,
         dependencies_description=None,
         env=[],
         arbitrary_files=None,
@@ -69,6 +72,7 @@ def __init__(
         self.input_files = input_files
         self.client_outputs = client_outputs or ClientOutputs()
         self.working_directory = working_directory
+        self.metadata_directory = metadata_directory
         self.dependencies_description = dependencies_description
         self.env = env
         self.rewrite_paths = rewrite_paths
@@ -103,9 +107,11 @@ def __init__(
         output_files=[],
         work_dir_outputs=None,
         version_file=None,
-        dynamic_outputs=None
+        dynamic_outputs=None,
+        metadata_directory=None,
     ):
         self.working_directory = working_directory
+        self.metadata_directory = metadata_directory
         self.work_dir_outputs = work_dir_outputs or []
         self.output_files = output_files or []
         self.version_file = version_file
@@ -115,6 +121,7 @@ def __init__(
     def to_dict(self):
         return dict(
             working_directory=self.working_directory,
+            metadata_directory=self.metadata_directory,
             work_dir_outputs=self.work_dir_outputs,
             output_files=self.output_files,
             version_file=self.version_file,
@@ -125,6 +132,7 @@ def to_dict(self):
     def from_dict(config_dict):
         return ClientOutputs(
             working_directory=config_dict.get('working_directory'),
+            metadata_directory=config_dict.get('metadata_directory'),
             work_dir_outputs=config_dict.get('work_dir_outputs'),
             output_files=config_dict.get('output_files'),
             version_file=config_dict.get('version_file'),
@@ -139,9 +147,12 @@ class PulsarOutputs(object):
     """ Abstraction describing the output files PRODUCED by the remote Pulsar
     server. """
 
-    def __init__(self, working_directory_contents, output_directory_contents, remote_separator=sep):
+    def __init__(
+        self, working_directory_contents, output_directory_contents, metadata_directory_contents, remote_separator=sep
+    ):
         self.working_directory_contents = working_directory_contents
         self.output_directory_contents = output_directory_contents
+        self.metadata_directory_contents = metadata_directory_contents
         self.path_helper = PathHelper(remote_separator)
 
     @staticmethod
@@ -150,13 +161,15 @@ def from_status_response(complete_response):
         # by the Pulsar - older Pulsar instances will not set these in complete response.
         working_directory_contents = complete_response.get("working_directory_contents")
         output_directory_contents = complete_response.get("outputs_directory_contents")
+        metadata_directory_contents = complete_response.get("metadata_directory_contents")
         # Older (pre-2014) Pulsar servers will not include separator in response,
         # so this should only be used when reasoning about outputs in
         # subdirectories (which was not previously supported prior to that).
         remote_separator = complete_response.get("system_properties", {}).get("separator", sep)
         return PulsarOutputs(
             working_directory_contents,
             output_directory_contents,
+            metadata_directory_contents,
             remote_separator
         )
 

pulsar/client/staging/down.py

@@ -1,6 +1,5 @@
 from os.path import join
 from os.path import relpath
-from re import compile
 from contextlib import contextmanager
 
 from ..staging import COMMAND_VERSION_FILENAME
@@ -10,12 +9,6 @@
 from logging import getLogger
 log = getLogger(__name__)
 
-# All output files marked with from_work_dir attributes will copied or downloaded
-# this pattern picks up attiditional files to copy back - such as those
-# associated with multiple outputs and metadata configuration. Set to .* to just
-# copy everything
-COPY_FROM_WORKING_DIRECTORY_PATTERN = compile(r"primary_.*|galaxy.json|metadata_.*|dataset_\d+\.dat|__instrument_.*|dataset_\d+_files.+")
-
 
 def finish_job(client, cleanup_job, job_completed_normally, client_outputs, pulsar_outputs):
     """ Responsible for downloading results from remote server and cleaning up
@@ -63,12 +56,14 @@ def __init__(self, output_collector, action_mapper, client_outputs, pulsar_outpu
         self.exception_tracker = DownloadExceptionTracker()
         self.output_files = client_outputs.output_files
         self.working_directory_contents = pulsar_outputs.working_directory_contents or []
+        self.metadata_directory_contents = pulsar_outputs.metadata_directory_contents or []
 
     def collect(self):
         self.__collect_working_directory_outputs()
         self.__collect_outputs()
         self.__collect_version_file()
         self.__collect_other_working_directory_files()
+        self.__collect_metadata_directory_files()
         return self.exception_tracker.collection_failure_exceptions
 
     def __collect_working_directory_outputs(self):
@@ -105,15 +100,31 @@ def __collect_version_file(self):
             self._attempt_collect_output('output', version_file, name=COMMAND_VERSION_FILENAME)
 
     def __collect_other_working_directory_files(self):
-        working_directory = self.client_outputs.working_directory
+        self.__collect_directory_files(
+            self.client_outputs.working_directory,
+            self.working_directory_contents,
+            'output_workdir',
+        )
+
+    def __collect_metadata_directory_files(self):
+        self.__collect_directory_files(
+            self.client_outputs.metadata_directory,
+            self.metadata_directory_contents,
+            'output_metadata',
+        )
+
+    def __collect_directory_files(self, directory, contents, output_type):
+        if directory is None:  # e.g. output_metadata_directory
+            return
+
         # Fetch remaining working directory outputs of interest.
-        for name in self.working_directory_contents:
+        for name in contents:
             if name in self.downloaded_working_directory_files:
                 continue
             if self.client_outputs.dynamic_match(name):
-                log.debug("collecting dynamic output %s" % name)
-                output_file = join(working_directory, self.pulsar_outputs.path_helper.local_name(name))
-                if self._attempt_collect_output(output_type='output_workdir', path=output_file, name=name):
+                log.debug("collecting dynamic %s file %s" % (output_type, name))
+                output_file = join(directory, self.pulsar_outputs.path_helper.local_name(name))
+                if self._attempt_collect_output(output_type=output_type, path=output_file, name=name):
                     self.downloaded_working_directory_files.append(name)
 
     def _attempt_collect_output(self, output_type, path, name=None):

pulsar/client/staging/up.py

@@ -74,6 +74,7 @@ def __init__(self, client, client_job_description, job_config):
             self.tool_version = None
             self.tool_dir = None
         self.working_directory = client_job_description.working_directory
+        self.metadata_directory = client_job_description.metadata_directory
         self.version_file = client_job_description.version_file
         self.arbitrary_files = client_job_description.arbitrary_files
         self.rewrite_paths = client_job_description.rewrite_paths
@@ -95,6 +96,7 @@ def __init__(self, client, client_job_description, job_config):
         self.__upload_tool_files()
         self.__upload_input_files()
         self.__upload_working_directory_files()
+        self.__upload_metadata_directory_files()
         self.__upload_arbitrary_files()
 
         if self.rewrite_paths:
@@ -190,12 +192,24 @@ def __upload_working_directory_files(self):
             path = join(self.working_directory, working_directory_file)
             self.transfer_tracker.handle_transfer(path, path_type.WORKDIR)
 
+    def __upload_metadata_directory_files(self):
+        metadata_directory_files = self.__metadata_directory_files()
+        for metadata_directory_file in metadata_directory_files:
+            path = join(self.metadata_directory, metadata_directory_file)
+            self.transfer_tracker.handle_transfer(path, path_type.METADATA)
+
     def __working_directory_files(self):
         if self.working_directory and exists(self.working_directory):
             return listdir(self.working_directory)
         else:
             return []
 
+    def __metadata_directory_files(self):
+        if self.metadata_directory and exists(self.metadata_directory):
+            return listdir(self.metadata_directory)
+        else:
+            return []
+
     def __initialize_version_file_rename(self):
         version_file = self.version_file
         if version_file: