Merge pull request #132 from common-workflow-language/curoverse-cwltool-collect-output-checksum-flag

Curoverse cwltool collect output checksum flag

Author: mr-c

cwltool/draft2tool.py

@@ -108,7 +108,9 @@ def __init__(self, job, output_callback, cachebuilder, jobcache):
     def run(self, **kwargs):
         # type: (**Any) -> None
         self.output_callback(self.job.collect_output_ports(self.job.tool["outputs"],
-                                                           self.cachebuilder, self.outdir),
+                                                           self.cachebuilder,
+                                                           self.outdir,
+                                                           kwargs.get("compute_checksum", True)),
                                             "success")
 
 # map files to assigned path inside a container. We need to also explicitly
@@ -337,12 +339,12 @@ def rm_pending_output_callback(output_callback, jobcachepending,
 
         j.pathmapper = builder.pathmapper
         j.collect_outputs = partial(
-                self.collect_output_ports, self.tool["outputs"], builder)
+                self.collect_output_ports, self.tool["outputs"], builder, compute_checksum=kwargs.get("compute_checksum", True))
         j.output_callback = output_callback
 
         yield j
 
-    def collect_output_ports(self, ports, builder, outdir):
+    def collect_output_ports(self, ports, builder, outdir, compute_checksum=True):
         # type: (Set[Dict[str,Any]], Builder, str) -> Dict[unicode, Union[unicode, List[Any], Dict[unicode, Any]]]
         try:
             ret = {}  # type: Dict[unicode, Union[unicode, List[Any], Dict[unicode, Any]]]
@@ -364,7 +366,7 @@ def collect_output_ports(self, ports, builder, outdir):
             for port in ports:
                 fragment = shortname(port["id"])
                 try:
-                    ret[fragment] = self.collect_output(port, builder, outdir)
+                    ret[fragment] = self.collect_output(port, builder, outdir, compute_checksum)
                 except Exception as e:
                     _logger.debug(u"Error collecting output for parameter '%s'" % shortname(port["id"]), exc_info=e)
                     raise WorkflowException(u"Error collecting output for parameter '%s': %s" % (shortname(port["id"]), e))
@@ -377,7 +379,7 @@ def collect_output_ports(self, ports, builder, outdir):
         except validate.ValidationException as e:
             raise WorkflowException("Error validating output record, " + str(e) + "\n in " + json.dumps(ret, indent=4))
 
-    def collect_output(self, schema, builder, outdir):
+    def collect_output(self, schema, builder, outdir, compute_checksum=True):
         # type: (Dict[str,Any], Builder, str) -> Union[Dict[unicode, Any], List[Union[Dict[unicode, Any], unicode]]]
         r = []  # type: List[Any]
         if "outputBinding" in schema:
@@ -410,17 +412,20 @@ def collect_output(self, schema, builder, outdir):
                     if files["class"] == "Directory" and "listing" not in files:
                         getListing(builder.fs_access, files)
                     else:
-                        checksum = hashlib.sha1()
                         with builder.fs_access.open(files["location"], "rb") as f:
-                            contents = f.read(CONTENT_LIMIT)
+                            contents = ""
+                            if binding.get("loadContents") or compute_checksum:
+                                contents = f.read(CONTENT_LIMIT)
                             if binding.get("loadContents"):
                                 files["contents"] = contents
-                            filesize = 0
-                            while contents != "":
-                                checksum.update(contents)
-                                filesize += len(contents)
-                                contents = f.read(1024*1024)
-                        files["checksum"] = "sha1$%s" % checksum.hexdigest()
+                            if compute_checksum:
+                                checksum = hashlib.sha1()
+                                while contents != "":
+                                    checksum.update(contents)
+                                    contents = f.read(1024*1024)
+                                files["checksum"] = "sha1$%s" % checksum.hexdigest()
+                            f.seek(0, 2)
+                            filesize = f.tell()
                         files["size"] = filesize
                         if "format" in schema:
                             files["format"] = builder.do_eval(schema["format"], context=files)
@@ -478,6 +483,6 @@ def collect_output(self, schema, builder, outdir):
             out = {}
             for f in schema["type"]["fields"]:
                 out[shortname(f["name"])] = self.collect_output(  # type: ignore
-                        f, builder, outdir)
+                        f, builder, outdir, compute_checksum)
             return out
         return r

cwltool/main.py

@@ -158,6 +158,14 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
                         help="Desired workflow behavior when a step fails.  One of 'stop' or 'continue'. "
                         "Default is 'stop.", default="stop")
 
+    exgroup = parser.add_mutually_exclusive_group()
+    exgroup.add_argument("--compute-checksum", action="store_true", default=True,
+                        help="Compute checksum of contents while collecting outputs",
+                        dest="compute_checksum")
+    exgroup.add_argument("--no-compute-checksum", action="store_false",
+                        help="Do not compute checksum of contents while collecting outputs",
+                        dest="compute_checksum")
+
     parser.add_argument("workflow", type=str, nargs="?", default=None)
     parser.add_argument("job_order", nargs=argparse.REMAINDER)