3 from functools import partial
7 from cStringIO import StringIO
9 import cwltool.draft2tool
10 from cwltool.draft2tool import CommandLineTool
11 import cwltool.workflow
12 from cwltool.process import get_feature, scandeps, UnsupportedRequirement, normalizeFilesDirs, shortname
13 from cwltool.load_tool import fetch_document
14 from cwltool.pathmapper import adjustFileObjs, adjustDirObjs
15 from cwltool.utils import aslist
16 from cwltool.builder import substitute
18 import arvados.collection
19 import ruamel.yaml as yaml
21 from .arvdocker import arv_docker_get_image
22 from .pathmapper import ArvPathMapper
23 from ._version import __version__
25 logger = logging.getLogger('arvados.cwl-runner')
27 cwltool.draft2tool.ACCEPTLIST_RE = re.compile(r".*")
29 def trim_listing(obj):
30 """Remove 'listing' field from Directory objects that are keep references.
32 When Directory objects represent Keep references, it redundant and
33 potentially very expensive to pass fully enumerated Directory objects
34 between instances of cwl-runner (e.g. a submitting a job, or using the
35 RunInSingleContainer feature), so delete the 'listing' field when it is
39 if obj.get("location", "").startswith("keep:") and "listing" in obj:
41 if obj.get("location", "").startswith("_:"):
44 def upload_dependencies(arvrunner, name, document_loader,
45 workflowobj, uri, loadref_run):
46 """Upload the dependencies of the workflowobj document to Keep.
48 Returns a pathmapper object mapping local paths to keep references. Also
49 does an in-place update of references in "workflowobj".
51 Use scandeps to find $import, $include, $schemas, run, File and Directory
52 fields that represent external references.
54 If workflowobj has an "id" field, this will reload the document to ensure
55 it is scanning the raw document prior to preprocessing.
60 joined = urlparse.urljoin(b, u)
61 defrg, _ = urlparse.urldefrag(joined)
62 if defrg not in loaded:
64 # Use fetch_text to get raw file (before preprocessing).
65 text = document_loader.fetch_text(defrg)
66 if isinstance(text, bytes):
67 textIO = StringIO(text.decode('utf-8'))
69 textIO = StringIO(text)
70 return yaml.safe_load(textIO)
75 loadref_fields = set(("$import", "run"))
77 loadref_fields = set(("$import",))
80 if "id" in workflowobj:
81 # Need raw file content (before preprocessing) to ensure
82 # that external references in $include and $mixin are captured.
83 scanobj = loadref("", workflowobj["id"])
85 sc = scandeps(uri, scanobj,
87 set(("$include", "$schemas", "location")),
90 normalizeFilesDirs(sc)
92 if "id" in workflowobj:
93 sc.append({"class": "File", "location": workflowobj["id"]})
95 mapper = ArvPathMapper(arvrunner, sc, "",
101 if "location" in p and (not p["location"].startswith("_:")) and (not p["location"].startswith("keep:")):
102 p["location"] = mapper.mapper(p["location"]).resolved
103 adjustFileObjs(workflowobj, setloc)
104 adjustDirObjs(workflowobj, setloc)
109 def upload_docker(arvrunner, tool):
110 if isinstance(tool, CommandLineTool):
111 (docker_req, docker_is_req) = get_feature(tool, "DockerRequirement")
113 if docker_req.get("dockerOutputDirectory"):
114 # TODO: can be supported by containers API, but not jobs API.
115 raise UnsupportedRequirement("Option 'dockerOutputDirectory' of DockerRequirement not supported.")
116 arv_docker_get_image(arvrunner.api, docker_req, True, arvrunner.project_uuid)
117 elif isinstance(tool, cwltool.workflow.Workflow):
119 upload_docker(arvrunner, s.embedded_tool)
121 def upload_instance(arvrunner, name, tool, job_order):
122 upload_docker(arvrunner, tool)
124 for t in tool.tool["inputs"]:
125 def setSecondary(fileobj):
126 if isinstance(fileobj, dict) and fileobj.get("class") == "File":
127 if "secondaryFiles" not in fileobj:
128 fileobj["secondaryFiles"] = [{"location": substitute(fileobj["location"], sf), "class": "File"} for sf in t["secondaryFiles"]]
130 if isinstance(fileobj, list):
134 if shortname(t["id"]) in job_order and t.get("secondaryFiles"):
135 setSecondary(job_order[shortname(t["id"])])
137 workflowmapper = upload_dependencies(arvrunner,
143 jobmapper = upload_dependencies(arvrunner,
144 os.path.basename(job_order.get("id", "#")),
147 job_order.get("id", "#"),
150 if "id" in job_order:
153 return workflowmapper
155 def arvados_jobs_image(arvrunner):
156 img = "arvados/jobs:"+__version__
158 arv_docker_get_image(arvrunner.api, {"dockerPull": img}, True, arvrunner.project_uuid)
159 except Exception as e:
160 raise Exception("Docker image %s is not available\n%s" % (img, e) )
163 class Runner(object):
164 def __init__(self, runner, tool, job_order, enable_reuse, output_name):
165 self.arvrunner = runner
167 self.job_order = job_order
169 self.enable_reuse = enable_reuse
171 self.final_output = None
172 self.output_name = output_name
174 def update_pipeline_component(self, record):
177 def arvados_job_spec(self, *args, **kwargs):
178 self.name = os.path.basename(self.tool.tool["id"])
179 workflowmapper = upload_instance(self.arvrunner, self.name, self.tool, self.job_order)
180 adjustDirObjs(self.job_order, trim_listing)
181 return workflowmapper
183 def done(self, record):
184 if record["state"] == "Complete":
185 if record.get("exit_code") is not None:
186 if record["exit_code"] == 33:
187 processStatus = "UnsupportedRequirement"
188 elif record["exit_code"] == 0:
189 processStatus = "success"
191 processStatus = "permanentFail"
193 processStatus = "success"
195 processStatus = "permanentFail"
200 self.final_output = record["output"]
201 outc = arvados.collection.CollectionReader(self.final_output,
202 api_client=self.arvrunner.api,
203 keep_client=self.arvrunner.keep_client,
204 num_retries=self.arvrunner.num_retries)
205 with outc.open("cwl.output.json") as f:
206 outputs = json.load(f)
207 def keepify(fileobj):
208 path = fileobj["location"]
209 if not path.startswith("keep:"):
210 fileobj["location"] = "keep:%s/%s" % (record["output"], path)
211 adjustFileObjs(outputs, keepify)
212 adjustDirObjs(outputs, keepify)
213 except Exception as e:
214 logger.error("While getting final output object: %s", e)
215 self.arvrunner.output_callback(outputs, processStatus)
217 del self.arvrunner.processes[record["uuid"]]