Warning
This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.tool_util.output_checker
import os
import re
import stat
from collections.abc import Iterable
from enum import Enum
from logging import getLogger
from pathlib import Path
from typing import (
Literal,
TYPE_CHECKING,
)
from typing_extensions import (
NotRequired,
TypedDict,
)
from galaxy.tool_util.parser.stdio import StdioErrorLevel
from galaxy.tool_util.runtime_environment import RUNTIME_ENVIRONMENT_WARNING_FILE
if TYPE_CHECKING:
from galaxy.tool_util.parser.stdio import (
ToolStdioExitCode,
ToolStdioRegex,
)
log = getLogger(__name__)
[docs]
class DETECTED_JOB_STATE(str, Enum):
OK = "ok"
OUT_OF_MEMORY_ERROR = "oom_error"
GENERIC_ERROR = "generic_error"
ERROR_PEEK_SIZE = 2000
JobMessageTypeLiteral = Literal[
"regex",
"exit_code",
"max_discovered_files",
"output_collection_security",
"output_discovery",
"stdio_read_error",
"unknown_datatype",
"runtime_environment_warning",
]
[docs]
class JobMessage(TypedDict):
desc: str | None
code_desc: NotRequired[str | None]
error_level: float # Literal[0, 1, 1.1, 2, 3, 4] - mypy doesn't like literal floats.
[docs]
class RegexJobMessage(JobMessage):
type: Literal["regex"]
stream: str | None
match: str | None
[docs]
class OutputCollectionSecurityJobMessage(JobMessage):
type: Literal["output_collection_security"]
[docs]
class StdioReadErrorJobMessage(JobMessage):
type: Literal["stdio_read_error"]
stream: str
errno: int | None
[docs]
class UnknownDatatypeJobMessage(JobMessage):
type: Literal["unknown_datatype"]
extension: str
[docs]
class RuntimeEnvironmentWarningJobMessage(JobMessage):
type: Literal["runtime_environment_warning"]
variable_names: list[str]
AnyJobMessage = (
ExitCodeJobMessage
| RegexJobMessage
| MaxDiscoveredFilesJobMessage
| OutputCollectionSecurityJobMessage
| OutputDiscoveryJobMessage
| StdioReadErrorJobMessage
| UnknownDatatypeJobMessage
| RuntimeEnvironmentWarningJobMessage
)
[docs]
def output_discovery_job_message(reason: str | None = None) -> OutputDiscoveryJobMessage:
desc = "Failed to collect job outputs"
if reason:
desc = f"{desc}: {reason}"
return OutputDiscoveryJobMessage(
type="output_discovery",
desc=desc,
code_desc=None,
error_level=StdioErrorLevel.FATAL,
)
[docs]
def unknown_datatype_job_message(extension: str, count: int) -> UnknownDatatypeJobMessage:
datasets = "1 output dataset" if count == 1 else f"{count} output datasets"
return UnknownDatatypeJobMessage(
type="unknown_datatype",
desc=f"Extension '{extension}' is not a datatype on this Galaxy server, {datasets} set to 'data'",
code_desc=None,
error_level=StdioErrorLevel.WARNING,
extension=extension,
)
[docs]
def check_output_regex(
regex: "ToolStdioRegex", stream: str, stream_name: str, job_messages: list[AnyJobMessage], max_error_level: int
) -> int:
"""
check a single regex against a stream
regex the regex to check
stream the stream to search in
job_messages a list where the descriptions of the detected regexes can be appended
max_error_level the maximum error level that has been detected so far
returns the max of the error_level of the regex and the given max_error_level
"""
regex_match = re.search(regex.match, stream, re.IGNORECASE)
if regex_match:
reason = __regex_err_msg(regex_match, stream_name, regex)
job_messages.append(reason)
return max(max_error_level, regex.error_level)
return max_error_level
[docs]
def check_output(
stdio_regexes: list["ToolStdioRegex"],
stdio_exit_codes: list["ToolStdioExitCode"],
stdout: str,
stderr: str,
tool_exit_code: int,
) -> tuple[str, str, str, list[AnyJobMessage]]:
"""
Check the output of a tool - given the stdout, stderr, and the tool's
exit code, return DETECTED_JOB_STATE.OK if the tool exited successfully or
error type otherwise. No exceptions should be thrown. If this code encounters
an exception, it returns OK so that the workflow can continue;
otherwise, a bug in this code could halt workflow progress.
Note that, if the tool did not define any exit code handling or
any stdio/stderr handling, then it reverts back to previous behavior:
if stderr contains anything, then False is returned.
"""
# By default, the tool succeeded. This covers the case where the code
# has a bug but the tool was ok, and it lets a workflow continue.
state = DETECTED_JOB_STATE.OK
# messages (descriptions of the detected exit_code and regexes)
# to be prepended to the stdout/stderr after all exit code and regex tests
# are done (otherwise added messages are searched again).
# messages are added it the order of detection
# If job is failed, track why.
job_messages: list[AnyJobMessage] = []
try:
# Check exit codes and match regular expressions against stdout and
# stderr if this tool was configured to do so.
# If there is a regular expression for scanning stdout/stderr,
# then we assume that the tool writer overwrote the default
# behavior of just setting an error if there is *anything* on
# stderr.
if len(stdio_regexes) > 0 or len(stdio_exit_codes) > 0:
# Check the exit code ranges in the order in which
# they were specified. Each exit_code is a StdioExitCode
# that includes an applicable range. If the exit code was in
# that range, then apply the error level and add a message.
# If we've reached a fatal error rule, then stop.
max_error_level = StdioErrorLevel.NO_ERROR
if tool_exit_code is not None:
for stdio_exit_code in stdio_exit_codes:
if tool_exit_code >= stdio_exit_code.range_start and tool_exit_code <= stdio_exit_code.range_end:
# Tack on a generic description of the code
# plus a specific code description. For example,
# this might prepend "Job 42: Warning (Out of Memory)\n".
code_desc = stdio_exit_code.desc
if None is code_desc:
code_desc = ""
desc = f"{StdioErrorLevel.desc(stdio_exit_code.error_level)}: Exit code {tool_exit_code} ({code_desc})"
reason: ExitCodeJobMessage = {
"type": "exit_code",
"desc": desc,
"exit_code": tool_exit_code,
"code_desc": code_desc,
"error_level": stdio_exit_code.error_level,
}
job_messages.append(reason)
max_error_level = max(max_error_level, stdio_exit_code.error_level)
if max_error_level >= StdioErrorLevel.MAX:
break
if max_error_level < StdioErrorLevel.FATAL_OOM:
# We'll examine every regex. Each regex specifies whether
# it is to be run on stdout, stderr, or both. (It is
# possible for neither stdout nor stderr to be scanned,
# but those regexes won't be used.) We record the highest
# error level, which are currently "warning" and "fatal".
# If fatal, then we set the job's state to ERROR.
# If warning, then we still set the job's state to OK
# but include a message. We'll do this if we haven't seen
# a fatal error yet
for regex in stdio_regexes:
# If ( this regex should be matched against stdout )
# - Run the regex's match pattern against stdout
# - If it matched, then determine the error level.
# o If it was fatal, then we're done - break.
if regex.stderr_match:
max_error_level = check_output_regex(regex, stderr, "stderr", job_messages, max_error_level)
if max_error_level >= StdioErrorLevel.MAX:
break
if regex.stdout_match:
max_error_level = check_output_regex(regex, stdout, "stdout", job_messages, max_error_level)
if max_error_level >= StdioErrorLevel.MAX:
break
# If we encountered a fatal error, then we'll need to set the
# job state accordingly. Otherwise the job is ok:
if max_error_level == StdioErrorLevel.FATAL_OOM:
state = DETECTED_JOB_STATE.OUT_OF_MEMORY_ERROR
elif max_error_level >= StdioErrorLevel.FATAL:
error_reason = ""
if job_messages:
error_reason = f" Reasons are {job_messages}"
log.info(f"Job error detected, failing job.{error_reason}")
state = DETECTED_JOB_STATE.GENERIC_ERROR
# When there are no regular expressions and no exit codes to check,
# default to the previous behavior: when there's anything on stderr
# the job has an error, and the job is ok otherwise.
else:
# TODO: Add in the tool and job id:
# log.debug( "Tool did not define exit code or stdio handling; "
# + "checking stderr for success" )
if stderr:
state = DETECTED_JOB_STATE.GENERIC_ERROR
peek = stderr[0:ERROR_PEEK_SIZE] if stderr else ""
log.info(f"Job failed because of contents in the standard error stream: [{peek}]")
except Exception:
log.exception("Job state check encountered unexpected exception; assuming execution successful")
return state, stdout, stderr, job_messages
def __regex_err_msg(match: re.Match, stream: str, regex: "ToolStdioRegex") -> RegexJobMessage:
"""
Return a message about the match on tool output using the given
ToolStdioRegex regex object. The regex_match is a MatchObject
that will contain the string matched on.
"""
# Get the description for the error level:
desc = f"{StdioErrorLevel.desc(regex.error_level)}: "
mstart = match.start()
mend = match.end()
if mend - mstart > 256:
match_str = f"{match.string[mstart : mstart + 256]}..."
else:
match_str = match.string[mstart:mend]
# If there's a description for the regular expression, then use it.
# Otherwise, we'll take the first 256 characters of the match.
if regex.desc is not None:
desc += match.expand(regex.desc)
else:
desc += f"Matched on {match_str}"
return {
"type": "regex",
"stream": stream,
"desc": desc,
"code_desc": regex.desc,
"match": match_str,
"error_level": regex.error_level,
}
# The warning file holds only variable names; anything larger is not ours.
MAX_RUNTIME_ENVIRONMENT_WARNING_BYTES = 64 * 1024
def _read_runtime_environment_warning_file(path: Path) -> str:
"""Read a file the tool could have replaced: no symlinks, regular files only, bounded size."""
try:
link_stat = os.lstat(path)
if not stat.S_ISREG(link_stat.st_mode):
return ""
fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
except OSError:
return ""
with os.fdopen(fd, "rb") as warning_file:
# Without O_NOFOLLOW the path could be swapped for a symlink after lstat.
if not os.path.samestat(link_stat, os.fstat(warning_file.fileno())):
return ""
return warning_file.read(MAX_RUNTIME_ENVIRONMENT_WARNING_BYTES).decode("utf-8", errors="replace")
def _runtime_environment_variable_names(job_directory: str, required_names: Iterable[str]) -> list[str]:
"""Read the recorded unset names, keeping only names the tool declares as required."""
required = set(required_names)
if not required:
return []
names: list[str] = []
for directory in ("outputs", "metadata"):
contents = _read_runtime_environment_warning_file(
Path(job_directory) / directory / RUNTIME_ENVIRONMENT_WARNING_FILE
)
names.extend(name for name in contents.splitlines() if name in required)
return list(dict.fromkeys(names))
[docs]
def merge_runtime_environment_warnings(
job_directory: str, task_directories: Iterable[str], required_names: Iterable[str]
) -> None:
"""Publish task warnings before the parent job's metadata and output checks."""
required = list(required_names)
names = list(
dict.fromkeys(
name for directory in task_directories for name in _runtime_environment_variable_names(directory, required)
)
)
path = Path(job_directory) / "outputs" / RUNTIME_ENVIRONMENT_WARNING_FILE
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("".join(f"{name}\n" for name in names))
[docs]
def runtime_environment_job_messages(
job_directory: str, required_names: Iterable[str]
) -> list[RuntimeEnvironmentWarningJobMessage]:
names = _runtime_environment_variable_names(job_directory, required_names)
if not names:
return []
desc = f"Required runtime environment variables are unset: {', '.join(names)}"
log.info(desc)
return [
RuntimeEnvironmentWarningJobMessage(
type="runtime_environment_warning", variable_names=names, desc=desc, error_level=StdioErrorLevel.WARNING
)
]