Feature - stdout live reporting

client/src/api/schema/schema.ts

@@ -2838,6 +2838,28 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/jobs/{job_id}/console_output": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /**
+         * Returns STDOUT and STDERR from the tool running in a specific job.
+         * @description Get the stdout and/or stderr from the tool running in a specific job. The position parameters are the index
+         *     of where to start reading stdout/stderr. The length parameters control how much
+         *     stdout/stderr is read.
+         */
+        get: operations["get_console_output_api_jobs__job_id__console_output_get"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/jobs/{job_id}/destination_params": {
         parameters: {
             query?: never;
@@ -12654,6 +12676,24 @@ export interface components {
              */
             update_time: string;
         };
+        /** JobConsoleOutput */
+        JobConsoleOutput: {
+            /**
+             * Job State
+             * @description The current job's state
+             */
+            state?: components["schemas"]["JobState"] | null;
+            /**
+             * STDERR
+             * @description Tool STDERR from job.
+             */
+            stderr?: string | null;
+            /**
+             * STDOUT
+             * @description Tool STDOUT from job.
+             */
+            stdout?: string | null;
+        };
         /** JobDestinationParams */
         JobDestinationParams: {
             /**
@@ -27871,6 +27911,54 @@ export interface operations {
             };
         };
     };
+    get_console_output_api_jobs__job_id__console_output_get: {
+        parameters: {
+            query: {
+                stdout_position: number;
+                stdout_length: number;
+                stderr_position: number;
+                stderr_length: number;
+            };
+            header?: {
+                /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+                "run-as"?: string | null;
+            };
+            path: {
+                job_id: string;
+            };
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["JobConsoleOutput"];
+                };
+            };
+            /** @description Request Error */
+            "4XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+            /** @description Server Error */
+            "5XX": {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MessageExceptionModel"];
+                };
+            };
+        };
+    };
     destination_params_job_api_jobs__job_id__destination_params_get: {
         parameters: {
             query?: never;

client/src/components/JobInformation/CodeRow.vue

@@ -1,10 +1,5 @@
 <template>
-    <tr
-        v-b-tooltip.hover
-        :title="`click to ${action}`"
-        @mousedown="mouseIsDown = true"
-        @mousemove="mouseIsDown ? (mouseMoved = true) : (mouseMoved = false)"
-        @mouseup="toggleExpanded()">
+    <tr>
         <td>
             <span v-if="helpUri">
                 <HelpText :uri="helpUri" :text="codeLabel" />
@@ -18,7 +13,13 @@
                 <b-col cols="11">
                     <pre :class="codeClass">{{ codeItem }}</pre>
                 </b-col>
-                <b-col class="nopadding pointer">
+                <b-col
+                    v-b-tooltip.hover
+                    class="nopadding pointer"
+                    :title="`click to ${action}`"
+                    @mousedown="mouseIsDown = true"
+                    @mousemove="mouseIsDown ? (mouseMoved = true) : (mouseMoved = false)"
+                    @mouseup="toggleExpanded()">
                     <FontAwesomeIcon :icon="iconClass" />
                 </b-col>
             </b-row>
@@ -48,6 +49,7 @@ export default {
             mouseIsDown: false,
             mouseMoved: false,
             expanded: false,
+            lastPos: 0,
         };
     },
     computed: {
@@ -61,6 +63,18 @@ export default {
             return this.expanded ? ["fas", "compress-alt"] : ["fas", "expand-alt"];
         },
     },
+    updated() {
+        try {
+            var codeDiv = this.$el.querySelector(".code");
+            if (codeDiv.scrollTop + codeDiv.offsetHeight >= this.lastPos - 5) {
+                // scroll is at the bottom
+                codeDiv.scrollTop = codeDiv.scrollHeight;
+            }
+            this.lastPos = codeDiv.scrollHeight;
+        } catch (exception) {
+            console.debug("Code div is not present");
+        }
+    },
     methods: {
         toggleExpanded() {
             this.mouseIsDown = false;
@@ -80,4 +94,8 @@ export default {
     padding: 0;
     margin: 0;
 }
+.code {
+    max-height: 50em;
+    overflow: auto;
+}
 </style>

client/src/components/JobInformation/JobInformation.vue

@@ -1,7 +1,7 @@
 <script setup>
 import CopyToClipboard from "components/CopyToClipboard";
 import HelpText from "components/Help/HelpText";
-import { JobDetailsProvider } from "components/providers/JobProvider";
+import { JobConsoleOutputProvider, JobDetailsProvider } from "components/providers/JobProvider";
 import UtcDate from "components/UtcDate";
 import { NON_TERMINAL_STATES } from "components/WorkflowInvocationState/util";
 import { computed, ref, watch } from "vue";
@@ -28,9 +28,21 @@ const props = defineProps({
     },
 });
 
+const stdout_length = ref(50000);
+const stdout_text = ref("");
+const stderr_length = ref(50000);
+const stderr_text = ref("");
+
+const stdout_position = computed(() => stdout_text.value.length);
+const stderr_position = computed(() => stderr_text.value.length);
+
 const runTime = computed(() => getJobDuration(job.value));
 
-const jobIsTerminal = computed(() => job.value && !NON_TERMINAL_STATES.includes(job.value.state));
+function jobStateIsTerminal(jobState) {
+    return jobState && !NON_TERMINAL_STATES.includes(job.value.state);
+}
+
+const jobIsTerminal = computed(() => jobStateIsTerminal(job?.value?.state));
 
 const routeToInvocation = computed(() => `/workflows/invocations/${invocationId.value}`);
 
@@ -41,6 +53,26 @@ const metadataDetail = ref({
 
 function updateJob(newJob) {
     job.value = newJob;
+    if (jobStateIsTerminal(newJob?.state)) {
+        if (newJob.tool_stdout) {
+            stdout_text.value = newJob.tool_stdout;
+        }
+        if (newJob.tool_stderr) {
+            stderr_text.value = newJob.tool_stderr;
+        }
+    }
+}
+
+function updateConsoleOutputs(output) {
+    // Keep stdout in memory and only fetch new text via JobProvider
+    if (output) {
+        if (output.stdout != null) {
+            stdout_text.value += output.stdout;
+        }
+        if (output.stderr != null) {
+            stderr_text.value += output.stderr;
+        }
+    }
 }
 
 function filterMetadata(jobMessages) {
@@ -94,6 +126,14 @@ watch(
 <template>
     <div>
         <JobDetailsProvider auto-refresh :job-id="props.job_id" @update:result="updateJob" />
+        <JobConsoleOutputProvider
+            auto-refresh
+            :job-id="props.job_id"
+            :stdout_position="stdout_position"
+            :stdout_length="stdout_length"
+            :stderr_position="stderr_position"
+            :stderr_length="stderr_length"
+            @update:result="updateConsoleOutputs" />
         <h2 class="h-md">Job Information</h2>
         <table id="job-information" class="tabletip info_data_table">
             <tbody>
@@ -146,13 +186,13 @@ watch(
                     id="stdout"
                     help-uri="unix.stdout"
                     :code-label="'Tool Standard Output'"
-                    :code-item="job.tool_stdout" />
+                    :code-item="stdout_text" />
                 <CodeRow
                     v-if="job"
                     id="stderr"
                     help-uri="unix.stderr"
                     :code-label="'Tool Standard Error'"
-                    :code-item="job.tool_stderr" />
+                    :code-item="stderr_text" />
                 <CodeRow
                     v-if="job && job.traceback"
                     id="traceback"
@@ -211,7 +251,6 @@ watch(
         </table>
     </div>
 </template>
-
 <style scoped>
 .tooltipJobInfo {
     text-decoration-line: underline;

client/src/components/providers/JobProvider.js

@@ -15,7 +15,26 @@ async function jobDetails({ jobId }) {
     }
 }
 
+async function jobConsoleOutput({
+    jobId,
+    stdout_position = 0,
+    stdout_length = 0,
+    stderr_position = 0,
+    stderr_length = 0,
+}) {
+    const url =
+        `${getAppRoot()}api/jobs/${jobId}/console_output?stdout_position=${stdout_position}&stdout_length=${stdout_length}` +
+        `&stderr_position=${stderr_position}&stderr_length=${stderr_length}`;
+    try {
+        const { data } = await axios.get(url);
+        return data;
+    } catch (e) {
+        rethrowSimple(e);
+    }
+}
+
 export const JobDetailsProvider = SingleQueryProvider(jobDetails, stateIsTerminal);
+export const JobConsoleOutputProvider = SingleQueryProvider(jobConsoleOutput, stateIsTerminal);
 
 export function jobsProvider(ctx, callback, extraParams = {}) {
     const { root, ...requestParams } = ctx;

lib/galaxy/jobs/runners/pulsar.py

@@ -652,6 +652,18 @@ def finish_job(self, job_state: JobState):
             remote_metadata_directory = run_results.get("metadata_directory", None)
             tool_stdout = unicodify(run_results.get("stdout", ""), strip_null=True)
             tool_stderr = unicodify(run_results.get("stderr", ""), strip_null=True)
+            for file in ("tool_stdout", "tool_stderr"):
+                if tool_stdout and tool_stderr:
+                    pass
+                try:
+                    file_path = os.path.join(job_wrapper.working_directory, "outputs", file)
+                    file_content = open(file_path)
+                    if tool_stdout is None and file == "tool_stdout":
+                        tool_stdout = file_content.read()
+                    elif tool_stderr is None and file == "tool_stderr":
+                        tool_stderr = file_content.read()
+                except Exception:
+                    pass
             job_stdout = run_results.get("job_stdout")
             job_stderr = run_results.get("job_stderr")
             exit_code = run_results.get("returncode")

lib/galaxy/managers/jobs.py

@@ -4,6 +4,7 @@
     date,
     datetime,
 )
+from pathlib import Path
 from typing import (
     Any,
     cast,
@@ -33,6 +34,7 @@
 
 from galaxy import model
 from galaxy.exceptions import (
+    ConfigDoesNotAllowException,
     ItemAccessibilityException,
     ObjectNotFound,
     RequestParameterInvalidException,
@@ -83,6 +85,7 @@
     defaultdict,
     ExecutionTimer,
     listify,
+    string_as_bool_or_none,
 )
 from galaxy.util.search import (
     FilteredTerm,
@@ -96,6 +99,10 @@
 JobStatesT = Union[JobStateT, List[JobStateT]]
 
 
+STDOUT_LOCATION = "outputs/tool_stdout"
+STDERR_LOCATION = "outputs/tool_stderr"
+
+
 class JobLock(BaseModel):
     active: bool = Field(title="Job lock status", description="If active, jobs will not dispatch")
 
@@ -296,6 +303,47 @@ def get_accessible_job(self, trans: ProvidesUserContext, decoded_job_id) -> Job:
         trans.sa_session.refresh(job)
         return job
 
+    def get_job_console_output(
+        self, trans, job, stdout_position=-1, stdout_length=0, stderr_position=-1, stderr_length=0
+    ):
+        if job is None:
+            raise ObjectNotFound()
+
+        # Check job destination params to see if stdout reporting is enabled
+        dest_params = job.destination_params
+        if not string_as_bool_or_none(dest_params.get("live_tool_output_reporting", False)):
+            raise ConfigDoesNotAllowException()
+
+        # If stdout_length and stdout_position are good values, then load standard out and add it to status
+        console_output = {}
+        console_output["state"] = job.state
+        if job.state == job.states.RUNNING:
+            working_directory = trans.app.object_store.get_filename(
+                job, base_dir="job_work", dir_only=True, obj_dir=True
+            )
+            if stdout_length > -1 and stdout_position > -1:
+                try:
+                    stdout_path = Path(working_directory) / STDOUT_LOCATION
+                    stdout_file = open(stdout_path)
+                    stdout_file.seek(stdout_position)
+                    console_output["stdout"] = stdout_file.read(stdout_length)
+                except Exception as e:
+                    log.error("Could not read STDOUT: %s", e)
+                    console_output["stdout"] = ""
+            if stderr_length > -1 and stderr_position > -1:
+                try:
+                    stderr_path = Path(working_directory) / STDERR_LOCATION
+                    stderr_file = open(stderr_path)
+                    stderr_file.seek(stderr_position)
+                    console_output["stderr"] = stderr_file.read(stderr_length)
+                except Exception as e:
+                    log.error("Could not read STDERR: %s", e)
+                    console_output["stderr"] = ""
+        else:
+            console_output["stdout"] = job.tool_stdout
+            console_output["stderr"] = job.tool_stderr
+        return console_output
+
     def stop(self, job, message=None):
         if not job.finished:
             job.mark_deleted(self.app.config.track_jobs_in_database)