Fix pulsar with metadata_strategy == 'directory'.

lib/galaxy/jobs/command_factory.py

@@ -30,7 +30,7 @@ def build_command(
     include_work_dir_outputs=True,
     create_tool_working_directory=True,
     remote_command_params={},
-    metadata_directory=None,
+    remote_job_directory=None,
     stdout_file=None,
     stderr_file=None,
 ):
@@ -111,8 +111,8 @@ def build_command(
     commands_builder.capture_return_code()
 
     if include_metadata and job_wrapper.requires_setting_metadata:
-        metadata_directory = metadata_directory or job_wrapper.working_directory
-        commands_builder.append_command("cd '%s'" % metadata_directory)
+        working_directory = remote_job_directory or job_wrapper.working_directory
+        commands_builder.append_command("cd '%s'" % working_directory)
         __handle_metadata(commands_builder, job_wrapper, runner, remote_command_params)
 
     return commands_builder.build()

lib/galaxy/jobs/runners/pulsar.py

@@ -286,12 +286,18 @@ class PulsarJobRunner(AsynchronousJobRunner):
                 unstructured_path_rewrites = compute_environment.unstructured_path_rewrites
                 output_names = compute_environment.output_names()
 
+            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_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.tool_working_directory,
-                metadata_directory=job_wrapper.working_directory,
+                metadata_directory=metadata_directory,
                 tool=job_wrapper.tool,
                 config_files=job_wrapper.extra_filenames,
                 dependencies_description=dependencies_description,
@@ -363,9 +369,6 @@ class PulsarJobRunner(AsynchronousJobRunner):
                 compute_job_directory=remote_job_directory,
             )
             job_wrapper.disable_commands_in_new_shell()
-            metadata_directory = None
-            if remote_metadata:
-                metadata_directory = remote_job_config['metadata_directory']
 
             # Pulsar handles ``create_tool_working_directory`` and
             # ``include_work_dir_outputs`` details.
@@ -374,10 +377,10 @@ class PulsarJobRunner(AsynchronousJobRunner):
                 job_wrapper=job_wrapper,
                 container=container,
                 include_metadata=remote_metadata,
-                metadata_directory=metadata_directory,
                 create_tool_working_directory=False,
                 include_work_dir_outputs=False,
                 remote_command_params=remote_command_params,
+                remote_job_directory=remote_job_directory,
             )
         except UnsupportedPulsarException:
             log.exception("failure running job %d, unsupported Pulsar target", job_wrapper.job_id)

test/integration/test_pulsar_embedded_metadata.py

@@ -0,0 +1,23 @@
+"""Integration tests for the Pulsar embedded runner."""
+
+import os
+
+from base import integration_util
+
+SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
+EMBEDDED_PULSAR_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "embedded_pulsar_metadata_job_conf.xml")
+
+
+class EmbeddedPulsarIntegrationInstance(integration_util.IntegrationInstance):
+    """Describe a Galaxy test instance with embedded pulsar configured."""
+
+    framework_tool_and_types = True
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["job_config_file"] = EMBEDDED_PULSAR_JOB_CONFIG_FILE
+
+
+instance = integration_util.integration_module_instance(EmbeddedPulsarIntegrationInstance)
+
+test_tools = integration_util.integration_tool_runner(["simple_constructs"])