Merge pull request #862 from common-workflow-language/prov-nested-wf-fixes

Provenance: PROV links for nested workflows

Author: stain

cwltool/executors.py

@@ -110,7 +110,7 @@ def execute(self,
                 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)
+                process.parent_wf.finalize_prov_profile(name=None)
             return (self.final_output[0], self.final_status[0])
         return (None, "permanentFail")
 
@@ -130,10 +130,12 @@ def run_jobs(self,
         # define provenance profile for single commandline tool
         if not isinstance(process, Workflow) \
                 and runtime_context.research_obj is not None:
-            orcid = runtime_context.orcid
-            full_name = runtime_context.cwl_full_name
             process.provenance_object = CreateProvProfile(
-                runtime_context.research_obj, orcid, full_name)
+                runtime_context.research_obj,
+                full_name=runtime_context.cwl_full_name,
+                orcid=runtime_context.orcid,
+                # single tool execution, so RO UUID = wf UUID = tool UUID
+                run_uuid=runtime_context.research_obj.ro_uuid)
             process.parent_wf = process.provenance_object
         jobiter = process.job(job_order_object, self.output_callback,
                               runtime_context)

cwltool/job.py

@@ -226,6 +226,7 @@ def _execute(self,
         if self.joborder and runtimeContext.research_obj:
             job_order = self.joborder
             assert runtimeContext.prov_obj
+            assert runtimeContext.process_run_id
             runtimeContext.prov_obj.used_artefacts(
                 job_order, runtimeContext.process_run_id, str(self.name))
         outputs = {}  # type: Dict[Text,Text]

cwltool/provenance.py

@@ -18,6 +18,7 @@
 import six
 from six import string_types
 from six.moves import range
+from uuid import UUID # pylint: disable=unused-import
 
 from . import command_line_tool, expression
 from .builder import CONTENT_LIMIT
@@ -255,7 +256,10 @@ def do_output_callback(self, final_output_callback):
             self.prov_obj.document.wasEndedBy(
                 self.prov_obj.workflow_run_uri, None, self.prov_obj.engine_uuid,
                 datetime.datetime.now())
-            self.prov_obj.finalize_prov_profile(str(self.name))
+            prov_ids = self.prov_obj.finalize_prov_profile(self.name)
+            # Tell parent to associate our provenance files with our wf run
+            self.parent_wf.activity_has_provenance(self.prov_obj.workflow_run_uri, prov_ids)
+
         _logger.info(u"[%s] completed %s", self.name, self.processStatus)
         if _logger.isEnabledFor(logging.DEBUG):
             _logger.debug(u"[%s] %s", self.name, json_dumps(wo, indent=4))
@@ -499,12 +503,23 @@ def __init__(self,
             toolpath_object, loadingContext)
         self.provenance_object = None  # type: Optional[CreateProvProfile]
         if loadingContext.research_obj:
-            orcid = loadingContext.orcid
-            full_name = loadingContext.cwl_full_name
+            run_uuid = None # type: Optional[UUID]
+            is_master = not(loadingContext.prov_obj) # Not yet set
+            if is_master:
+                run_uuid = loadingContext.research_obj.ro_uuid
+
             self.provenance_object = CreateProvProfile(
-                loadingContext.research_obj, full_name, orcid,
-                loadingContext.host_provenance, loadingContext.user_provenance)
+                loadingContext.research_obj,
+                full_name=loadingContext.cwl_full_name,
+                orcid=loadingContext.orcid,
+                host_provenance=loadingContext.host_provenance,
+                user_provenance=loadingContext.user_provenance,
+                run_uuid=run_uuid # inherit RO UUID for master wf run
+                )
+            # TODO: Is Workflow(..) only called when we are the master workflow?
             self.parent_wf = self.provenance_object
+
+        # FIXME: Won't this overwrite prov_obj for nested workflows?
         loadingContext.prov_obj = self.provenance_object
         loadingContext = loadingContext.copy()
         loadingContext.requirements = self.requirements