From 74a661548867576d39ccc3fa676578a61fd45327 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Mon, 8 Apr 2019 15:14:27 -0400 Subject: [PATCH] Fix pulsar with metadata_strategy == 'directory'. --- lib/galaxy/jobs/command_factory.py | 6 ++--- lib/galaxy/jobs/runners/pulsar.py | 13 +++++++---- .../test_pulsar_embedded_metadata.py | 23 +++++++++++++++++++ 3 files changed, 34 insertions(+), 8 deletions(-) create mode 100644 test/integration/test_pulsar_embedded_metadata.py diff --git a/lib/galaxy/jobs/command_factory.py b/lib/galaxy/jobs/command_factory.py index 99ad26a6443..ad99b0e1e56 100644 --- a/lib/galaxy/jobs/command_factory.py +++ b/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() diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py index 1f9d2ccc656..0e0f5c89b80 100644 --- a/lib/galaxy/jobs/runners/pulsar.py +++ b/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) diff --git a/test/integration/test_pulsar_embedded_metadata.py b/test/integration/test_pulsar_embedded_metadata.py new file mode 100644 index 00000000000..d5b5582dbcb --- /dev/null +++ b/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"])