Track job failure reasons in a more structured way.

lib/galaxy/jobs/output_checker.py

@@ -14,22 +14,23 @@ DETECTED_JOB_STATE = Bunch(
     GENERIC_ERROR='generic_error',
 )
 
+ERROR_PEAK = 2000
 
-def check_output_regex(job, regex, stream, stream_append, max_error_level):
+
+def check_output_regex(job, regex, stream, stream_name, job_messages, max_error_level):
     """
     check a single regex against a stream
 
     regex the regex to check
     stream the stream to search in
-    stream_append a list where the descriptions of the detected regexes can be appended
+    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:
-        rexmsg = __regex_err_msg(regex_match, regex)
-        log.info("Job %s: %s" % (job.get_id_tag(), rexmsg))
-        stream_append.append(rexmsg)
+        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
 
@@ -51,9 +52,6 @@ def check_output_regex_byline(job, regex, stream, stream_append, max_error_level
     return max_error_level
 
 
-ERROR_PEAK = 2000
-
-
 def check_output(tool, stdout, stderr, tool_exit_code, job):
     """
     Check the output of a tool - given the stdout, stderr, and the tool's
@@ -77,8 +75,9 @@ def check_output(tool, stdout, stderr, tool_exit_code, job):
     # 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
-    stderr_toolmsg = []
-    stdout_toolmsg = []
+
+    # If job is failed, track why.
+    job_messages = []
 
     try:
         # Check exit codes and match regular expressions against stdout and
@@ -104,12 +103,19 @@ def check_output(tool, stdout, stderr, tool_exit_code, job):
                         code_desc = stdio_exit_code.desc
                         if None is code_desc:
                             code_desc = ""
-                        tool_msg = ("%s: Exit code %d (%s)" % (
+                        desc = "%s: Exit code %d (%s)" % (
                             StdioErrorLevel.desc(stdio_exit_code.error_level),
                             tool_exit_code,
-                            code_desc))
-                        log.info("Job %s: %s" % (job.get_id_tag(), tool_msg))
-                        stderr_toolmsg.append(tool_msg)
+                            code_desc)
+                        reason = {
+                            'type': 'exit_code',
+                            'desc': desc,
+                            'exit_code': tool_exit_code,
+                            'code_desc': code_desc,
+                            'error_level': stdio_exit_code.error_level,
+                        }
+                        log.info("Job %s: %s" % (job.get_id_tag(), reason))
+                        job_messages.append(reason)
                         max_error_level = max(max_error_level,
                                               stdio_exit_code.error_level)
                         if max_error_level >= StdioErrorLevel.MAX:
@@ -130,15 +136,13 @@ def check_output(tool, stdout, stderr, tool_exit_code, job):
                     #   - 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.
-                    # Repeat the stdout stuff for stderr.
-                    # TODO could test for stderr first? Reason: I would expect it to be smaller and contain the errors
-                    if regex.stdout_match:
-                        max_error_level = check_output_regex_byline(job, regex, stdout, stdout_toolmsg, max_error_level)
+                    if regex.stderr_match:
+                        max_error_level = check_output_regex(job, regex, stderr, 'stderr', job_messages, max_error_level)
                         if max_error_level >= StdioErrorLevel.MAX:
                             break
 
-                    if regex.stderr_match:
-                        max_error_level = check_output_regex_byline(job, regex, stderr, stderr_toolmsg, max_error_level)
+                    if regex.stdout_match:
+                        max_error_level = check_output_regex(job, regex, stdout, 'stdout', job_messages, max_error_level)
                         if max_error_level >= StdioErrorLevel.MAX:
                             break
 
@@ -177,35 +181,38 @@ def check_output(tool, stdout, stderr, tool_exit_code, job):
         state = DETECTED_JOB_STATE.OK
 
     # Store the modified stdout and stderr in the job:
-    if len(stdout_toolmsg) > 0:
-        stdout = "%s\n### END of messages added by Galaxy AND START of original stdout\n%s" % ("\n".join(stdout_toolmsg), stdout)
-    if len(stderr_toolmsg) > 0:
-        stderr = "%s\n### END of messages added by Galaxy AND START of original stderr\n%s" % ("\n".join(stderr_toolmsg), stderr)
     if job is not None:
-        job.set_streams(stdout, stderr)
+        job.set_streams(stdout, stderr, job_messages=job_messages)
 
     return state
 
 
-def __regex_err_msg(match, regex):
+def __regex_err_msg(match, stream, regex):
     """
     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:
-    err_msg = StdioErrorLevel.desc(regex.error_level) + ": "
+    desc = StdioErrorLevel.desc(regex.error_level) + ": "
+    mstart = match.start()
+    mend = match.end()
+    if mend - mstart > 256:
+        match_str = 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 None is not regex.desc:
-        err_msg += regex.desc
+    if regex.desc is not None:
+        desc += regex.desc
     else:
-        mstart = match.start()
-        mend = match.end()
-        err_msg += "Matched on "
-        # TODO: Move the constant 256 somewhere else besides here.
-        if mend - mstart > 256:
-            err_msg += match.string[mstart : mstart + 256] + "..."
-        else:
-            err_msg += match.string[mstart: mend]
-    return err_msg
+        desc += "Matched on %s" % match_str
+    return {
+        "type": "regex",
+        "stream": stream,
+        "desc": desc,
+        "code_desc": regex.desc,
+        "match": match_str,
+        "error_level": regex.error_level,
+    }

lib/galaxy/model/__init__.py

@@ -230,7 +230,7 @@ class JobLike(object):
         # TODO: Make iterable, concatenate with chain
         return self.text_metrics + self.numeric_metrics
 
-    def set_streams(self, stdout, stderr):
+    def set_streams(self, stdout, stderr, job_messages=None):
         stdout = galaxy.util.unicodify(stdout) or u''
         stderr = galaxy.util.unicodify(stderr) or u''
         if (len(stdout) > galaxy.util.DATABASE_MAX_STRING_SIZE):
@@ -241,6 +241,8 @@ class JobLike(object):
             stderr = galaxy.util.shrink_string_by_size(stderr, galaxy.util.DATABASE_MAX_STRING_SIZE, join_by="\n..\n", left_larger=True, beginning_on_size_error=True)
             log.info("stderr for %s %d is greater than %s, only a portion will be logged to database", type(self), self.id, galaxy.util.DATABASE_MAX_STRING_SIZE_PRETTY)
         self.stderr = stderr
+        if job_messages is not None:
+            self.job_messages = job_messages
 
     def log_str(self):
         extra = ""
@@ -607,6 +609,7 @@ class Job(JobLike, UsesCreateAndUpdateTime, Dictifiable, RepresentById):
         self.imported = False
         self.handler = None
         self.exit_code = None
+        self.job_messages = None
         self._init_metrics()
         self.state_history.append(JobStateHistory(self))
 

lib/galaxy/model/mapping.py

@@ -527,6 +527,7 @@ model.Job.table = Table(
     Column("copied_from_job_id", Integer, nullable=True),
     Column("command_line", TEXT),
     Column("dependencies", JSONType, nullable=True),
+    Column("job_messages", JSONType, nullable=True),
     Column("param_filename", String(1024)),
     Column("runner_name", String(255)),
     Column("stdout", TEXT),
@@ -724,6 +725,7 @@ model.Task.table = Table(
     Column("stdout", TEXT),
     Column("stderr", TEXT),
     Column("exit_code", Integer, nullable=True),
+    Column("job_messages", JSONType, nullable=True),
     Column("info", TrimmedString(255)),
     Column("traceback", TEXT),
     Column("job_id", Integer, ForeignKey("job.id"), index=True, nullable=False),

lib/galaxy/model/migrate/versions/0147_job_messages.py

@@ -0,0 +1,54 @@
+"""
+Add structured failure reason column to jobs table
+"""
+from __future__ import print_function
+
+import logging
+
+from sqlalchemy import Column, MetaData, Table
+
+from galaxy.model.custom_types import JSONType
+
+log = logging.getLogger(__name__)
+job_messages_column = Column("job_messages", JSONType, nullable=True)
+task_job_messages_column = Column("job_messages", JSONType, nullable=True)
+
+
+def upgrade(migrate_engine):
+    print(__doc__)
+    metadata = MetaData()
+    metadata.bind = migrate_engine
+    metadata.reflect()
+
+    try:
+        jobs_table = Table("job", metadata, autoload=True)
+        job_messages_column.create(jobs_table)
+        assert job_messages_column is jobs_table.c.job_messages
+    except Exception:
+        log.exception("Adding column 'job_messages' to job table failed.")
+
+    try:
+        tasks_table = Table("task", metadata, autoload=True)
+        task_job_messages_column.create(tasks_table)
+        assert task_job_messages_column is tasks_table.c.job_messages
+    except Exception:
+        log.exception("Adding column 'job_messages' to task table failed.")
+
+
+def downgrade(migrate_engine):
+    metadata = MetaData()
+    metadata.bind = migrate_engine
+    metadata.reflect()
+
+    try:
+        jobs_table = Table("job", metadata, autoload=True)
+        job_messages = jobs_table.c.job_messages
+        job_messages.drop()
+    except Exception:
+        log.exception("Dropping 'job_messages' column from job table failed.")
+    try:
+        tasks_table = Table("task", metadata, autoload=True)
+        job_messages = tasks_table.c.job_messages
+        job_messages.drop()
+    except Exception:
+        log.exception("Dropping 'job_messages' column from task table failed.")

lib/galaxy/tools/linters/tests.py

@@ -19,7 +19,7 @@ def lint_tsts(tool_xml, lint_ctx):
             has_test = True
         if len(test.findall("assert_stdout")) > 0:
             has_test = True
-        if len(test.findall("assert_stdout")) > 0:
+        if len(test.findall("assert_stderr")) > 0:
             has_test = True
         if len(test.findall("assert_command")) > 0:
             has_test = True

lib/galaxy/tools/verify/interactor.py

@@ -883,11 +883,36 @@ def _verify_outputs(testdef, history, jobs, tool_id, data_list, data_collection_
         "stdout": "Standard output of the job",
         "stderr": "Standard error of the job",
     }
+    # TODO: Only hack the stdio like this for older profkle, for newer tool profiles
+    # add some syntax for asserting job messages maybe - or just drop this because exit
+    # code and regex on stdio can be tested directly - so this is really testing Galaxy
+    # core handling more than the tool.
+    job_messages = job_stdio.get("job_messages") or []
+    stdout_prefix = ""
+    stderr_prefix = ""
+    for job_message in job_messages:
+        message_type = job_message.get("type")
+        if message_type == "regex" and job_message.get("stream") == "stderr":
+            stderr_prefix += (job_message.get("desc") or '') + "\n"
+        elif message_type == "regex" and job_message.get("stream") == "stdout":
+            stdout_prefix += (job_message.get("desc") or '') + "\n"
+        elif message_type == "exit_code":
+            stderr_prefix += (job_message.get("desc") or '') + "\n"
+        else:
+            raise Exception("Unknown job message type [%s] in [%s]" % (message_type, job_message))
+
     for what, description in other_checks.items():
         if getattr(testdef, what, None) is not None:
             try:
-                data = job_stdio[what]
-                verify_assertions(data, getattr(testdef, what))
+                raw_data = job_stdio[what]
+                assertions = getattr(testdef, what)
+                if what == "stdout":
+                    data = stdout_prefix + raw_data
+                elif what == "stderr":
+                    data = stderr_prefix + raw_data
+                else:
+                    data = raw_data
+                verify_assertions(data, assertions)
             except AssertionError as err:
                 errmsg = '%s different than expected\n' % description
                 errmsg += str(err)

lib/galaxy/webapps/galaxy/api/jobs.py

@@ -134,7 +134,7 @@ class JobController(BaseAPIController, UsesLibraryMixinItems):
         job_dict = self.encode_all_ids(trans, job.to_dict('element', system_details=is_admin), True)
         full_output = util.asbool(kwd.get('full', 'false'))
         if full_output:
-            job_dict.update(dict(stderr=job.stderr, stdout=job.stdout))
+            job_dict.update(dict(stderr=job.stderr, stdout=job.stdout, job_messages=job.job_messages))
             if is_admin:
                 if job.user:
                     job_dict['user_email'] = job.user.email

templates/show_params.mako

@@ -173,6 +173,13 @@
         <tr><td>Tool Standard Error:</td><td><a href="${h.url_for( controller='dataset', action='stderr', dataset_id=encoded_hda_id )}">stderr</a></td></tr>
         %if job:
             <tr><td>Tool Exit Code:</td><td>${ job.exit_code | h }</td></tr>
+            %if job.job_messages:
+            <tr><td>Job Messages</td><td><ul style="padding-left: 15px; margin-bottom: 0px">
+            %for job_message in job.job_messages:
+            <li>${ job_message['desc'] |h }</li>
+            %endfor
+            <ul></td></tr>
+            %endif
         %endif
         <tr><td>History Content API ID:</td>
         <td>${encoded_hda_id}

test/api/test_jobs.py

@@ -123,6 +123,12 @@ class JobsApiTestCase(api.ApiTestCase):
         job_details = show_jobs_response.json()
         self._assert_has_key(job_details, 'id', 'state', 'exit_code', 'update_time', 'create_time')
 
+        show_jobs_response = self._get("jobs/%s" % job_id, {"full": True})
+        self._assert_status_code_is(show_jobs_response, 200)
+
+        job_details = show_jobs_response.json()
+        self._assert_has_key(job_details, 'id', 'state', 'exit_code', 'update_time', 'create_time', 'stdout', 'stderr', 'job_messages')
+
     @uses_test_history(require_new=True)
     def test_show_security(self, history_id):
         self.__history_with_new_dataset(history_id)

test/functional/tools/samples_tool_conf.xml

@@ -48,6 +48,7 @@
   <tool file="metadata_column_names.xml" />
   <tool file="strict_shell.xml" />
   <tool file="strict_shell_default_off.xml" />
+  <tool file="detect_errors.xml" />
   <tool file="detect_errors_aggressive.xml" />
   <tool file="md5sum.xml" />
   <tool file="checksum.xml" />

test/unit/tools/test_parsing.py

@@ -485,6 +485,7 @@ class CollectionTestCase(BaseLoaderTestCase):
         tests = tests_dict["tests"]
         assert len(tests) == 2
         assert len(tests[0]["inputs"]) == 3, tests[0]
+
         outputs, output_collections = self._tool_source.parse_outputs(None)
         assert len(output_collections) == 0
 
@@ -505,3 +506,16 @@ class CollectionOutputYamlTestCase(BaseLoaderTestCase):
     def test_tests(self):
         outputs, output_collections = self._tool_source.parse_outputs(None)
         assert len(output_collections) == 1
+
+
+class ExpectationsTestCase(BaseLoaderTestCase):
+    source_file_name = os.path.join(os.getcwd(), "test/functional/tools/detect_errors.xml")
+    source_contents = None
+
+    def test_tests(self):
+        tests_dict = self._tool_source.parse_tests_to_dict()
+        tests = tests_dict["tests"]
+        assert len(tests) == 10
+        test_0 = tests[0]
+        assert len(test_0["stderr"]) == 1
+        assert len(test_0["stdout"]) == 2