Merge pull request #676 from common-workflow-language/provenance

Provenance Support for cwltool - Single Job Executor

Author: mr-c

.gitignore

@@ -30,6 +30,8 @@ typeshed/2and3/ruamel/yaml
 
 #mypy
 .mypy_cache/
+bin/
+lib/
 
 # Files generated by Makefile
 .cache/

CWLProv.rst

@@ -32,6 +32,13 @@ DEBDEVPKGS=pep8 python-autopep8 pylint python-coverage pydocstyle sloccount \
 VERSION=1.0.$(shell date +%Y%m%d%H%M%S --utc --date=`git log --first-parent \
 	--max-count=1 --format=format:%cI`)
 mkfile_dir := $(dir $(abspath $(lastword $(MAKEFILE_LIST))))
+UNAME_S=$(shell uname -s)
+ifeq ($(UNAME_S),Linux)
+	nproc=$(shell nproc)
+endif
+ifeq ($(UNAME_S),Darwin)
+	nproc=$(shell sysctl -n hw.physicalcpu)
+endif
 
 ## all         : default task
 all:
@@ -105,11 +112,11 @@ format: autopep8
 ## pylint      : run static code analysis on Python code
 pylint: $(PYSOURCES)
 	pylint --msg-template="{path}:{line}: [{msg_id}({symbol}), {obj}] {msg}" \
-                $^ -j$(shell nproc)|| true
+                $^ -j$(nproc)|| true
 
 pylint_report.txt: ${PYSOURCES}
 	pylint --msg-template="{path}:{line}: [{msg_id}({symbol}), {obj}] {msg}" \
-		$^ -j$(shell nproc)> pylint_report.txt || true
+		$^ -j$(nproc)> pylint_report.txt || true
 
 diff_pylint_report: pylint_report.txt
 	diff-quality --violations=pylint pylint_report.txt

Makefile

@@ -116,6 +116,43 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
                         type=float,
                         default=20)
 
+    provgroup = parser.add_argument_group("Options for recording provenance "
+                                          "information of the execution")
+    provgroup.add_argument("--provenance",
+                           help="Save provenance to specified folder as a "
+                           "Research Object that captures and aggregates "
+                           "workflow execution and data products.",
+                           type=Text)
+
+    provgroup.add_argument("--enable-user-provenance", default=False,
+                           action="store_true",
+                           help="Record user account info as part of provenance.",
+                           dest="user_provenance")
+    provgroup.add_argument("--disable-user-provenance", default=False,
+                           action="store_false",
+                           help="Do not record user account info in provenance.",
+                           dest="user_provenance")
+    provgroup.add_argument("--enable-host-provenance", default=False,
+                           action="store_true",
+                           help="Record host info as part of provenance.",
+                           dest="host_provenance")
+    provgroup.add_argument("--disable-host-provenance", default=False,
+                           action="store_false",
+                           help="Do not record host info in provenance.",
+                           dest="host_provenance")
+    provgroup.add_argument(
+        "--orcid", help="Record user ORCID identifier as part of "
+        "provenance, e.g. https://orcid.org/0000-0002-1825-0097 "
+        "or 0000-0002-1825-0097. Alternatively the environment variable "
+        "ORCID may be set.", dest="orcid", default=os.environ.get("ORCID"),
+        type=Text)
+    provgroup.add_argument(
+        "--full-name", help="Record full name of user as part of provenance, "
+        "e.g. Josiah Carberry. You may need to use shell quotes to preserve "
+        "spaces. Alternatively the environment variable CWL_FULL_NAME may "
+        "be set.", dest="cwl_full_name", default=os.environ.get("CWL_FULL_NAME"),
+        type=Text)
+
     exgroup = parser.add_mutually_exclusive_group()
     exgroup.add_argument("--print-rdf", action="store_true",
                          help="Print corresponding RDF graph for workflow and exit")

cwltool/argparser.py

@@ -3,7 +3,7 @@
 import copy
 import logging
 from typing import (Any, Callable, Dict, List,  # pylint: disable=unused-import
-                    Optional, Set, Text, Type, Union, Tuple)
+                    Optional, Set, Text, Type, Union, Tuple, TYPE_CHECKING)
 
 from rdflib import Graph, URIRef  # pylint: disable=unused-import
 from rdflib.namespace import OWL, RDFS
@@ -22,7 +22,8 @@
 from .stdfsaccess import StdFsAccess  # pylint: disable=unused-import
 from .utils import (aslist, docker_windows_path_adjust,
                     json_dumps, onWindows)
-
+if TYPE_CHECKING:
+    from .provenance import CreateProvProfile  # pylint: disable=unused-import
 CONTENT_LIMIT = 64 * 1024
 
 
@@ -178,6 +179,7 @@ def __init__(self,
 
         # One of "no_listing", "shallow_listing", "deep_listing"
         self.loadListing = loadListing
+        self.prov_obj = None  # type: Optional[CreateProvProfile]
 
         self.find_default_container = None  # type: Optional[Callable[[], Text]]
         self.job_script_provider = job_script_provider

cwltool/builder.py

@@ -11,7 +11,8 @@
 import tempfile
 from functools import cmp_to_key, partial
 from typing import (Any, Callable, Dict,  # pylint: disable=unused-import
-                    Generator, List, Optional, Set, Text, Type, Union, cast)
+                    Generator, List, Optional, Set, Text, Type, TYPE_CHECKING,
+                    Union, cast)
 
 import schema_salad.validate as validate
 from schema_salad.ref_resolver import file_uri, uri_file_path
@@ -40,7 +41,10 @@
 from .utils import (aslist, convert_pathsep_to_unix,
                     docker_windows_path_adjust, json_dumps, onWindows,
                     windows_default_container_id)
-from .context import LoadingContext, RuntimeContext, getdefault
+from .context import (LoadingContext,  # pylint: disable=unused-import
+                      RuntimeContext, getdefault)
+if TYPE_CHECKING:
+    from .provenance import CreateProvProfile  # pylint: disable=unused-import
 
 ACCEPTLIST_EN_STRICT_RE = re.compile(r"^[a-zA-Z0-9._+-]+$")
 ACCEPTLIST_EN_RELAXED_RE = re.compile(r".*")  # Accept anything
@@ -80,6 +84,7 @@ def __init__(self,
             self.outdir = outdir
             self.tmpdir = tmpdir
             self.script = script
+            self.prov_obj = None  # type: Optional[CreateProvProfile]
 
         def run(self, runtimeContext):  # type: (RuntimeContext) -> None
             try:
@@ -99,9 +104,11 @@ def job(self,
         # type: (...) -> Generator[ExpressionTool.ExpressionJob, None, None]
         builder = self._init_job(job_order, runtimeContext)
 
-        yield ExpressionTool.ExpressionJob(
+        job = ExpressionTool.ExpressionJob(
             builder, self.tool["expression"], output_callbacks,
             self.requirements, self.hints)
+        job.prov_obj = runtimeContext.prov_obj
+        yield job
 
 
 def remove_path(f):  # type: (Dict[Text, Any]) -> None
@@ -167,6 +174,7 @@ def __init__(self, job, output_callback, cachebuilder, jobcache):
         self.output_callback = output_callback
         self.cachebuilder = cachebuilder
         self.outdir = jobcache
+        self.prov_obj = None  # type: Optional[CreateProvProfile]
 
     def run(self, runtimeContext):
         # type: (RuntimeContext) -> None
@@ -213,8 +221,8 @@ def check_valid_locations(fs_access, ob):
 class CommandLineTool(Process):
     def __init__(self, toolpath_object, loadingContext):
         # type: (Dict[Text, Any], LoadingContext) -> None
-        super(CommandLineTool, self).__init__(
-            toolpath_object, loadingContext)
+        super(CommandLineTool, self).__init__(toolpath_object, loadingContext)
+        self.prov_obj = loadingContext.prov_obj
 
     def make_job_runner(self,
                         runtimeContext       # type: RuntimeContext
@@ -264,7 +272,7 @@ def job(self,
             job_order,         # type: Dict[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
             runtimeContext     # RuntimeContext
-            ):
+           ):
         # type: (...) -> Generator[Union[JobBase, CallbackJob], None, None]
 
         require_prefix = ""
@@ -369,6 +377,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
         j = self.make_job_runner(runtimeContext)(
             builder, builder.job, self.make_path_mapper, self.requirements,
             self.hints, jobname)
+        j.prov_obj = self.prov_obj
         j.successCodes = self.tool.get("successCodes")
         j.temporaryFailCodes = self.tool.get("temporaryFailCodes")
         j.permanentFailCodes = self.tool.get("permanentFailCodes")

cwltool/command_line_tool.py

@@ -15,6 +15,8 @@
 from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from .process import Process
+    from .provenance import (ResearchObject,  # pylint: disable=unused-import
+                             CreateProvProfile)
 
 class ContextBase(object):
     def __init__(self, kwargs=None):
@@ -50,6 +52,12 @@ def __init__(self, kwargs=None):
         self.resolver = None
         self.fetcher_constructor = None
         self.construct_tool_object = default_make_tool
+        self.research_obj = None           # type: Optional[ResearchObject]
+        self.orcid = None
+        self.cwl_full_name = None
+        self.host_provenance = False       # type: bool
+        self.user_provenance = False       # type: bool
+        self.prov_obj = None               # type: Optional[CreateProvProfile]
 
         super(LoadingContext, self).__init__(kwargs)
 
@@ -108,6 +116,12 @@ def __init__(self, kwargs=None):
         self.cidfile_dir = None
         self.cidfile_prefix = None
 
+        self.research_obj = None        # type: Optional[ResearchObject]
+        self.orcid = None
+        self.cwl_full_name = None
+        self.process_run_id = None      # type: Optional[str]
+        self.prov_obj = None            # type: Optional[CreateProvProfile]
+        self.reference_locations = {}   # type: Dict[Text, Text]
         super(RuntimeContext, self).__init__(kwargs)
 
 

cwltool/context.py

@@ -1,8 +1,8 @@
-import logging
 import os
 import tempfile
 import threading
 from abc import ABCMeta, abstractmethod
+import datetime
 from typing import (Any, Dict, List, Optional,  # pylint: disable=unused-import
                     Set, Text, Tuple)
 
@@ -12,14 +12,15 @@
 
 from .builder import Builder  # pylint: disable=unused-import
 from .errors import WorkflowException
+from .loghandler import _logger
 from .job import JobBase  # pylint: disable=unused-import
 from .mutation import MutationManager
+from .provenance import CreateProvProfile
 from .process import (Process,  # pylint: disable=unused-import
                       cleanIntermediate, relocateOutputs)
 from .utils import DEFAULT_TMP_PREFIX
-from .context import LoadingContext, RuntimeContext, getdefault
-
-_logger = logging.getLogger("cwltool")
+from .context import RuntimeContext, getdefault  # pylint: disable=unused-import
+from .workflow import Workflow, WorkflowJob, WorkflowJobStep
 
 class JobExecutor(six.with_metaclass(ABCMeta, object)):
     """ Abstract base job executor. """
@@ -92,6 +93,18 @@ def execute(self,
             cleanIntermediate(self.output_dirs)
 
         if self.final_output and self.final_status:
+
+            if runtimeContext.research_obj is not None and \
+                    isinstance(process, (JobBase, Process, WorkflowJobStep,
+                                         WorkflowJob)) and process.parent_wf:
+                process_run_id = None
+                name = "primary"
+                process.parent_wf.generate_output_prov(self.final_output[0],
+                                                       process_run_id, name)
+                process.parent_wf.document.wasEndedBy(
+                    process.parent_wf.workflow_run_uri, None, process.parent_wf.engine_uuid,
+                    datetime.datetime.now())
+                process.parent_wf.finalize_prov_profile(name)
             return (self.final_output[0], self.final_status[0])
         return (None, "permanentFail")
 
@@ -104,6 +117,18 @@ def run_jobs(self,
                  logger,
                  runtimeContext     # type: RuntimeContext
                 ):  # type: (...) -> None
+
+        process_run_id = None  # type: Optional[str]
+        reference_locations = {}  # type: Dict[Text,Text]
+
+        # define provenance profile for single commandline tool
+        if not isinstance(process, Workflow) \
+                and runtimeContext.research_obj is not None:
+            orcid = runtimeContext.orcid
+            full_name = runtimeContext.cwl_full_name
+            process.provenance_object = CreateProvProfile(
+                runtimeContext.research_obj, orcid, full_name)
+            process.parent_wf = process.provenance_object
         jobiter = process.job(job_order_object, self.output_callback, runtimeContext)
 
         try:
@@ -113,6 +138,21 @@ def run_jobs(self,
                         job.builder = runtimeContext.builder
                     if job.outdir:
                         self.output_dirs.add(job.outdir)
+                    if runtimeContext.research_obj is not None:
+                        if not isinstance(process, Workflow):
+                            runtimeContext.prov_obj = process.provenance_object
+                        else:
+                            runtimeContext.prov_obj = job.prov_obj
+                        assert runtimeContext.prov_obj
+                        process_run_id, reference_locations = \
+                                runtimeContext.prov_obj.evaluate(
+                                        process, job, job_order_object,
+                                        runtimeContext.make_fs_access,
+                                        runtimeContext)
+                        runtimeContext = runtimeContext.copy()
+                        runtimeContext.process_run_id = process_run_id
+                        runtimeContext.reference_locations = \
+                            reference_locations
                     job.run(runtimeContext)
                 else:
                     logger.error("Workflow cannot make any more progress.")