Better errors (#234)

* Improve error reporting for expression evaluation.
* Connecting errors back to source file lines where possible.
* Better reporting on missing input files.
* Updating also should preserve line numbers as much as possible.

Author: tetron

cwltool/builder.py

@@ -3,6 +3,7 @@
 from . import expression
 import avro
 import schema_salad.validate as validate
+from schema_salad.sourceline import SourceLine
 from typing import Any, Callable, Text, Type, Union
 from .errors import WorkflowException
 from .stdfsaccess import StdFsAccess
@@ -36,6 +37,7 @@ def __init__(self):  # type: () -> None
         self.stagedir = None  # type: Text
         self.make_fs_access = None  # type: Type[StdFsAccess]
         self.build_job_script = None  # type: Callable[[List[str]], Text]
+        self.debug = False  # type: bool
 
     def bind_input(self, schema, datum, lead_pos=[], tail_pos=[]):
         # type: (Dict[Text, Any], Any, Union[int, List[int]], List[int]) -> List[Dict[Text, Any]]
@@ -155,7 +157,8 @@ def tostr(self, value):  # type: (Any) -> Text
     def generate_arg(self, binding):  # type: (Dict[Text,Any]) -> List[Text]
         value = binding.get("datum")
         if "valueFrom" in binding:
-            value = self.do_eval(binding["valueFrom"], context=value)
+            with SourceLine(binding, "valueFrom", WorkflowException):
+                value = self.do_eval(binding["valueFrom"], context=value)
 
         prefix = binding.get("prefix")
         sep = binding.get("separate", True)
@@ -203,4 +206,5 @@ def do_eval(self, ex, context=None, pull_image=True, recursive=False):
                                   self.outdir, self.tmpdir,
                                   self.resources,
                                   context=context, pull_image=pull_image,
-                                  timeout=self.timeout)
+                                  timeout=self.timeout,
+                                  debug=self.debug)

cwltool/cwlrdf.py

@@ -1,13 +1,13 @@
 import json
 import urlparse
 from .process import Process
-from schema_salad.ref_resolver import Loader
+from schema_salad.ref_resolver import Loader, ContextType
 from schema_salad.jsonld_context import makerdf
 from rdflib import Graph, plugin, URIRef
 from rdflib.serializer import Serializer
 from typing import Any, Dict, IO, Text, Union
 
-def gather(tool, ctx):  # type: (Process, Loader.ContextType) -> Graph
+def gather(tool, ctx):  # type: (Process, ContextType) -> Graph
     g = Graph()
 
     def visitor(t):
@@ -17,7 +17,7 @@ def visitor(t):
     return g
 
 def printrdf(wf, ctx, sr, stdout):
-    # type: (Process, Loader.ContextType, Text, IO[Any]) -> None
+    # type: (Process, ContextType, Text, IO[Any]) -> None
     stdout.write(gather(wf, ctx).serialize(format=sr))
 
 def lastpart(uri):  # type: (Any) -> Text
@@ -158,7 +158,7 @@ def dot_without_parameters(g, stdout):  # type: (Graph, IO[Any]) -> None
 
 
 def printdot(wf, ctx, stdout, include_parameters=False):
-    # type: (Process, Loader.ContextType, Any, bool) -> None
+    # type: (Process, ContextType, Any, bool) -> None
     g = gather(wf, ctx)
 
     stdout.write("digraph {")

cwltool/draft2tool.py

@@ -25,6 +25,8 @@
 from .job import CommandLineJob
 from .stdfsaccess import StdFsAccess
 
+from schema_salad.sourceline import SourceLine, indent
+
 ACCEPTLIST_EN_STRICT_RE = re.compile(r"^[a-zA-Z0-9._+-]+$")
 ACCEPTLIST_EN_RELAXED_RE = re.compile(r"^[ a-zA-Z0-9._+-]+$")  # with spaces
 ACCEPTLIST_RE = ACCEPTLIST_EN_STRICT_RE
@@ -159,11 +161,7 @@ def makeJobRunner(self):  # type: () -> CommandLineJob
     def makePathMapper(self, reffiles, stagedir, **kwargs):
         # type: (List[Any], Text, **Any) -> PathMapper
         dockerReq, _ = self.get_requirement("DockerRequirement")
-        try:
-            return PathMapper(reffiles, kwargs["basedir"], stagedir)
-        except OSError as e:
-            if e.errno == errno.ENOENT:
-                raise WorkflowException(u"Missing input file %s" % e)
+        return PathMapper(reffiles, kwargs["basedir"], stagedir)
 
     def job(self, joborder, output_callback, **kwargs):
         # type: (Dict[Text, Text], Callable[..., Any], **Any) -> Generator[Union[CommandLineJob, CallbackJob], None, None]
@@ -286,18 +284,21 @@ def rm_pending_output_callback(output_callback, jobcachepending,
         adjustDirObjs(builder.bindings, _check_adjust)
 
         if self.tool.get("stdin"):
-            j.stdin = builder.do_eval(self.tool["stdin"])
-            reffiles.append({"class": "File", "path": j.stdin})
+            with SourceLine(self.tool, "stdin", validate.ValidationException):
+                j.stdin = builder.do_eval(self.tool["stdin"])
+                reffiles.append({"class": "File", "path": j.stdin})
 
         if self.tool.get("stderr"):
-            j.stderr = builder.do_eval(self.tool["stderr"])
-            if os.path.isabs(j.stderr) or ".." in j.stderr:
-                raise validate.ValidationException("stderr must be a relative path")
+            with SourceLine(self.tool, "stderr", validate.ValidationException):
+                j.stderr = builder.do_eval(self.tool["stderr"])
+                if os.path.isabs(j.stderr) or ".." in j.stderr:
+                    raise validate.ValidationException("stderr must be a relative path, got '%s'" % j.stderr)
 
         if self.tool.get("stdout"):
-            j.stdout = builder.do_eval(self.tool["stdout"])
-            if os.path.isabs(j.stdout) or ".." in j.stdout or not j.stdout:
-                raise validate.ValidationException("stdout must be a relative path")
+            with SourceLine(self.tool, "stdout", validate.ValidationException):
+                j.stdout = builder.do_eval(self.tool["stdout"])
+                if os.path.isabs(j.stdout) or ".." in j.stdout or not j.stdout:
+                    raise validate.ValidationException("stdout must be a relative path, got '%s'" % j.stdout)
 
         if _logger.isEnabledFor(logging.DEBUG):
             _logger.debug(u"[job %s] command line bindings is %s", j.name, json.dumps(builder.bindings, indent=4))
@@ -389,17 +390,18 @@ def collect_output_ports(self, ports, builder, outdir, compute_checksum=True):
                 if _logger.isEnabledFor(logging.DEBUG):
                     _logger.debug(u"Raw output from %s: %s", custom_output, json.dumps(ret, indent=4))
             else:
-                for port in ports:
-                    fragment = shortname(port["id"])
-                    try:
-                        ret[fragment] = self.collect_output(port, builder, outdir, fs_access, compute_checksum=compute_checksum)
-                    except Exception as e:
-                        _logger.debug(
-                            u"Error collecting output for parameter '%s'"
-                            % shortname(port["id"]), exc_info=True)
-                        raise WorkflowException(
-                            u"Error collecting output for parameter '%s': %s"
-                            % (shortname(port["id"]), e))
+                for i, port in enumerate(ports):
+                    with SourceLine(ports, i, WorkflowException):
+                        fragment = shortname(port["id"])
+                        try:
+                            ret[fragment] = self.collect_output(port, builder, outdir, fs_access, compute_checksum=compute_checksum)
+                        except Exception as e:
+                            _logger.debug(
+                                u"Error collecting output for parameter '%s'"
+                                % shortname(port["id"]), exc_info=True)
+                            raise WorkflowException(
+                                u"Error collecting output for parameter '%s':\n%s"
+                                % (shortname(port["id"]), indent(unicode(e))))
 
             if ret:
                 adjustFileObjs(ret,
@@ -427,24 +429,25 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
             revmap = partial(revmap_file, builder, outdir)
 
             if "glob" in binding:
-                for gb in aslist(binding["glob"]):
-                    gb = builder.do_eval(gb)
-                    if gb:
-                        globpatterns.extend(aslist(gb))
-
-                for gb in globpatterns:
-                    if gb.startswith(outdir):
-                        gb = gb[len(outdir)+1:]
-                    elif gb == ".":
-                        gb = outdir
-                    elif gb.startswith("/"):
-                        raise WorkflowException("glob patterns must not start with '/'")
-                    try:
-                        r.extend([{"location": g,
-                                   "class": "File" if fs_access.isfile(g) else "Directory"}
-                                  for g in fs_access.glob(fs_access.join(outdir, gb))])
-                    except (OSError, IOError) as e:
-                        _logger.warn(Text(e))
+                with SourceLine(binding, "glob", WorkflowException):
+                    for gb in aslist(binding["glob"]):
+                        gb = builder.do_eval(gb)
+                        if gb:
+                            globpatterns.extend(aslist(gb))
+
+                    for gb in globpatterns:
+                        if gb.startswith(outdir):
+                            gb = gb[len(outdir)+1:]
+                        elif gb == ".":
+                            gb = outdir
+                        elif gb.startswith("/"):
+                            raise WorkflowException("glob patterns must not start with '/'")
+                        try:
+                            r.extend([{"location": g,
+                                       "class": "File" if fs_access.isfile(g) else "Directory"}
+                                      for g in fs_access.glob(fs_access.join(outdir, gb))])
+                        except (OSError, IOError) as e:
+                            _logger.warn(Text(e))
 
                 for files in r:
                     if files["class"] == "Directory" and "listing" not in files:
@@ -479,11 +482,13 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                 single = True
 
             if "outputEval" in binding:
-                r = builder.do_eval(binding["outputEval"], context=r)
+                with SourceLine(binding, "outputEval", WorkflowException):
+                    r = builder.do_eval(binding["outputEval"], context=r)
 
             if single:
                 if not r and not optional:
-                    raise WorkflowException("Did not find output file with glob pattern: '{}'".format(globpatterns))
+                    with SourceLine(binding, "glob", WorkflowException):
+                        raise WorkflowException("Did not find output file with glob pattern: '{}'".format(globpatterns))
                 elif not r and optional:
                     pass
                 elif isinstance(r, list):
@@ -498,20 +503,21 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                 Callable[[Any], Any], revmap))
 
             if "secondaryFiles" in schema:
-                for primary in aslist(r):
-                    if isinstance(primary, dict):
-                        primary["secondaryFiles"] = []
-                        for sf in aslist(schema["secondaryFiles"]):
-                            if isinstance(sf, dict) or "$(" in sf or "${" in sf:
-                                sfpath = builder.do_eval(sf, context=primary)
-                                if isinstance(sfpath, basestring):
-                                    sfpath = revmap({"location": sfpath, "class": "File"})
-                            else:
-                                sfpath = {"location": substitute(primary["location"], sf), "class": "File"}
-
-                            for sfitem in aslist(sfpath):
-                                if fs_access.exists(sfitem["location"]):
-                                    primary["secondaryFiles"].append(sfitem)
+                with SourceLine(schema, "secondaryFiles", WorkflowException):
+                    for primary in aslist(r):
+                        if isinstance(primary, dict):
+                            primary["secondaryFiles"] = []
+                            for sf in aslist(schema["secondaryFiles"]):
+                                if isinstance(sf, dict) or "$(" in sf or "${" in sf:
+                                    sfpath = builder.do_eval(sf, context=primary)
+                                    if isinstance(sfpath, basestring):
+                                        sfpath = revmap({"location": sfpath, "class": "File"})
+                                else:
+                                    sfpath = {"location": substitute(primary["location"], sf), "class": "File"}
+
+                                for sfitem in aslist(sfpath):
+                                    if fs_access.exists(sfitem["location"]):
+                                        primary["secondaryFiles"].append(sfitem)
 
             if not r and optional:
                 r = None

cwltool/expression.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import re
+import copy
 
 from typing import Any, AnyStr, Union, Text, Dict, List
 import schema_salad.validate as validate
@@ -106,33 +107,54 @@ def scanner(scan):  # type: (Text) -> List[int]
     else:
         return None
 
-def next_seg(remain, obj):  # type: (Text, Any)->Text
+def next_seg(remain, obj):  # type: (Text, Any) -> Any
     if remain:
         m = segment_re.match(remain)
+        key = None  # type: Union[str, int]
         if m.group(0)[0] == '.':
-            return next_seg(remain[m.end(0):], obj[m.group(0)[1:]])
+            key = m.group(0)[1:]
         elif m.group(0)[1] in ("'", '"'):
             key = m.group(0)[2:-2].replace("\\'", "'").replace('\\"', '"')
-            return next_seg(remain[m.end(0):], obj[key])
+
+        if key:
+            if isinstance(obj, list) and key == "length" and not remain[m.end(0):]:
+                return len(obj)
+            if not isinstance(obj, dict):
+                raise WorkflowException(" is a %s, cannot index on string '%s'" % (type(obj).__name__, key))
+            if key not in obj:
+                raise WorkflowException(" does not contain key '%s'" % key)
         else:
-            key = m.group(0)[1:-1]
-            return next_seg(remain[m.end(0):], obj[int(key)])
+            try:
+                key = int(m.group(0)[1:-1])
+            except ValueError as v:
+                raise WorkflowException(unicode(v))
+            if not isinstance(obj, list):
+                raise WorkflowException(" is a %s, cannot index on int '%s'" % (type(obj).__name__, key))
+            if key >= len(obj):
+                raise WorkflowException(" list index %i out of range" % key)
+        try:
+            return next_seg(remain[m.end(0):], obj[key])
+        except WorkflowException as w:
+            raise WorkflowException("%s%s" % (m.group(0), w))
     else:
         return obj
 
-def evaluator(ex, jslib, obj, fullJS=False, timeout=None):
-    # type: (Text, Text, Dict[Text, Any], bool, int) -> JSON
+def evaluator(ex, jslib, obj, fullJS=False, timeout=None, debug=False):
+    # type: (Text, Text, Dict[Text, Any], bool, int, bool) -> JSON
     m = param_re.match(ex)
     if m:
-        return next_seg(m.group(0)[m.end(1) - m.start(0):-1], obj[m.group(1)])
+        try:
+            return next_seg(m.group(0)[m.end(1) - m.start(0):-1], obj[m.group(1)])
+        except Exception as w:
+            raise WorkflowException("%s%s" % (m.group(1), w))
     elif fullJS:
-        return sandboxjs.execjs(ex, jslib, timeout=timeout)
+        return sandboxjs.execjs(ex, jslib, timeout=timeout, debug=debug)
     else:
         raise sandboxjs.JavascriptException("Syntax error in parameter reference '%s' or used Javascript code without specifying InlineJavascriptRequirement.", ex)
 
 def interpolate(scan, rootvars,
-                timeout=None, fullJS=None, jslib=""):
-    # type: (Text, Dict[Text, Any], int, bool, Union[str, Text]) -> JSON
+                timeout=None, fullJS=None, jslib="", debug=False):
+    # type: (Text, Dict[Text, Any], int, bool, Union[str, Text], bool) -> JSON
     scan = scan.strip()
     parts = []
     w = scanner(scan)
@@ -141,7 +163,7 @@ def interpolate(scan, rootvars,
 
         if scan[w[0]] == '$':
             e = evaluator(scan[w[0]+1:w[1]], jslib, rootvars, fullJS=fullJS,
-                          timeout=timeout)
+                          timeout=timeout, debug=debug)
             if w[0] == 0 and w[1] == len(scan):
                 return e
             leaf = json.dumps(e, sort_keys=True)
@@ -158,10 +180,10 @@ def interpolate(scan, rootvars,
     return ''.join(parts)
 
 def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
-            context=None, pull_image=True, timeout=None):
-    # type: (Union[dict, AnyStr], Dict[Text, Union[Dict, List, Text]], List[Dict[Text, Any]], Text, Text, Dict[Text, Union[int, Text]], Any, bool, int) -> Any
+            context=None, pull_image=True, timeout=None, debug=False):
+    # type: (Union[dict, AnyStr], Dict[Text, Union[Dict, List, Text]], List[Dict[Text, Any]], Text, Text, Dict[Text, Union[int, Text]], Any, bool, int, bool) -> Any
 
-    runtime = resources.copy()
+    runtime = copy.copy(resources)
     runtime["tmpdir"] = tmpdir
     runtime["outdir"] = outdir
 
@@ -179,10 +201,14 @@ def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
                 jslib = jshead(r.get("expressionLib", []), rootvars)
                 break
 
-        return interpolate(ex,
-                           rootvars,
-                           timeout=timeout,
-                           fullJS=fullJS,
-                           jslib=jslib)
+        try:
+            return interpolate(ex,
+                               rootvars,
+                               timeout=timeout,
+                               fullJS=fullJS,
+                               jslib=jslib,
+                               debug=debug)
+        except Exception as e:
+            raise WorkflowException("Expression evaluation error:\n%s" % e)
     else:
         return ex

cwltool/job.py

@@ -305,7 +305,7 @@ def linkoutdir(src, tgt):
                 _logger.exception("Exception while running job")
             processStatus = "permanentFail"
         except WorkflowException as e:
-            _logger.error(u"Error while running job: %s" % e)
+            _logger.error(u"[job %s] Job error:\n%s" % (self.name, e))
             processStatus = "permanentFail"
         except Exception as e:
             _logger.exception("Exception while running job")