Fix tool stdio on pulsar and implement job stdio

Depends on galaxyproject/pulsar#318
mvdbeek · Apr 6, 2023 · b89bf9e · b89bf9e
1 parent 0698979
commit b89bf9e
Show file tree

Hide file tree

Showing 5 changed files with 15 additions and 12 deletions.
diff --git a/lib/galaxy/jobs/command_factory.py b/lib/galaxy/jobs/command_factory.py
@@ -106,11 +106,10 @@ def build_command(
         else:
             commands_builder = CommandsBuilder(externalized_commands)
 
-    wrap_stdio = externalized or not for_pulsar
+    wrap_stdio = externalized and not for_pulsar
     if wrap_stdio:
-        # Galaxy writes I/O files to outputs, Pulsar uses metadata. metadata seems like
-        # it should be preferred - at least if the directory exists.
-        io_directory = "../metadata" if for_pulsar else "../outputs"
+        # Galaxy writes I/O files to outputs, Pulsar instruments own I/O redirection
+        io_directory = "../outputs"
         commands_builder.capture_stdout_stderr(
             f"{io_directory}/tool_stdout", f"{io_directory}/tool_stderr", stream_stdout_stderr=stream_stdout_stderr
         )

diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py
@@ -619,15 +619,16 @@ def finish_job(self, job_state: JobState):
         assert isinstance(
             job_state, AsynchronousJobState
         ), f"job_state type is '{type(job_state)}', expected AsynchronousJobState"
-        stderr = stdout = ""
         job_wrapper = job_state.job_wrapper
         try:
             client = self.get_client_from_state(job_state)
             run_results = client.full_status()
             remote_metadata_directory = run_results.get("metadata_directory", None)
-            stdout = run_results.get("stdout", "")
-            stderr = run_results.get("stderr", "")
-            exit_code = run_results.get("returncode", None)
+            tool_stdout = run_results.get("stdout", "")
+            tool_stderr = run_results.get("stderr", "")
+            job_stdout = run_results.get("job_stdout")
+            job_stderr = run_results.get("job_stderr")
+            exit_code = run_results.get("returncode")
             pulsar_outputs = PulsarOutputs.from_status_response(run_results)
             state = job_wrapper.get_state()
             # Use Pulsar client code to transfer/copy files back
@@ -668,9 +669,11 @@ def finish_job(self, job_state: JobState):
             ):
                 job_metrics_directory = job_wrapper.working_directory
             job_wrapper.finish(
-                stdout,
-                stderr,
+                tool_stdout,
+                tool_stderr,
                 exit_code,
+                job_stdout=job_stdout,
+                job_stderr=job_stderr,
                 remote_metadata_directory=remote_metadata_directory,
                 job_metrics_directory=job_metrics_directory,
             )

diff --git a/test/integration/test_pulsar_embedded.py b/test/integration/test_pulsar_embedded.py
@@ -31,6 +31,7 @@ def handle_galaxy_config_kwds(cls, config):
         "vcf_bgzip_test",
         "environment_variables",
         "multi_output_assign_primary_ext_dbkey",
+        "job_properties",
         "strict_shell",
         "tool_provided_metadata_9",
     ]

diff --git a/test/integration/test_pulsar_embedded_extended_metadata.py b/test/integration/test_pulsar_embedded_extended_metadata.py
@@ -29,7 +29,7 @@ def handle_galaxy_config_kwds(cls, config):
         "version_command_tool_dir",
         "simple_constructs",
         "metadata_bam",
-        # "job_properties",  # https://github.com/galaxyproject/galaxy/issues/11813
+        "job_properties",
         "from_work_dir_glob",
     ]
 )
diff --git a/test/integration/test_pulsar_embedded_remote_metadata.py b/test/integration/test_pulsar_embedded_remote_metadata.py
@@ -26,6 +26,6 @@ def handle_galaxy_config_kwds(cls, config):
     [
         "simple_constructs",
         "metadata_bam",
-        # "job_properties",  # https://github.com/galaxyproject/galaxy/issues/11813
+        "job_properties",
     ]
 )