Merge branch '18947-githttpd'
[arvados.git] / sdk / cwl / arvados_cwl / executor.py
index 9a94095ae8648d7107cfd1d4025098c117866802..680ca0b7b2c85df6b2f7d55709205b47ad591ef7 100644 (file)
@@ -18,6 +18,7 @@ import json
 import re
 from functools import partial
 import time
+import urllib
 
 from cwltool.errors import WorkflowException
 import cwltool.workflow
@@ -31,19 +32,18 @@ from arvados.errors import ApiError
 
 import arvados_cwl.util
 from .arvcontainer import RunnerContainer
-from .arvjob import RunnerJob, RunnerTemplate
 from .runner import Runner, upload_docker, upload_job_order, upload_workflow_deps
 from .arvtool import ArvadosCommandTool, validate_cluster_target, ArvadosExpressionTool
 from .arvworkflow import ArvadosWorkflow, upload_workflow
 from .fsaccess import CollectionFsAccess, CollectionFetcher, collectionResolver, CollectionCache, pdh_size
 from .perf import Perf
 from .pathmapper import NoFollowPathMapper
-from .task_queue import TaskQueue
+from cwltool.task_queue import TaskQueue
 from .context import ArvLoadingContext, ArvRuntimeContext
 from ._version import __version__
 
 from cwltool.process import shortname, UnsupportedRequirement, use_custom_schema
-from cwltool.pathmapper import adjustFileObjs, adjustDirObjs, get_listing, visit_class
+from cwltool.utils import adjustFileObjs, adjustDirObjs, get_listing, visit_class, aslist
 from cwltool.command_line_tool import compute_checksums
 from cwltool.load_tool import load_tool
 
@@ -91,8 +91,8 @@ class RuntimeStatusLoggingHandler(logging.Handler):
 
 
 class ArvCwlExecutor(object):
-    """Execute a CWL tool or workflow, submit work (using either jobs or
-    containers API), wait for them to complete, and report output.
+    """Execute a CWL tool or workflow, submit work (using containers API),
+    wait for them to complete, and report output.
 
     """
 
@@ -100,7 +100,8 @@ class ArvCwlExecutor(object):
                  arvargs=None,
                  keep_client=None,
                  num_retries=4,
-                 thread_count=4):
+                 thread_count=4,
+                 stdout=sys.stdout):
 
         if arvargs is None:
             arvargs = argparse.Namespace()
@@ -133,6 +134,7 @@ class ArvCwlExecutor(object):
         self.should_estimate_cache_size = True
         self.fs_access = None
         self.secret_store = None
+        self.stdout = stdout
 
         if keep_client is not None:
             self.keep_client = keep_client
@@ -154,7 +156,7 @@ class ArvCwlExecutor(object):
                                            num_retries=self.num_retries)
 
         self.work_api = None
-        expected_api = ["containers", "jobs"]
+        expected_api = ["containers"]
         for api in expected_api:
             try:
                 methods = self.api._rootDesc.get('resources')[api]['methods']
@@ -172,19 +174,11 @@ class ArvCwlExecutor(object):
                 raise Exception("Unsupported API '%s', expected one of %s" % (arvargs.work_api, expected_api))
 
         if self.work_api == "jobs":
-            logger.warning("""
+            logger.error("""
 *******************************
-Using the deprecated 'jobs' API.
-
-To get rid of this warning:
-
-Users: read about migrating at
-http://doc.arvados.org/user/cwl/cwl-style.html#migrate
-and use the option --api=containers
-
-Admins: configure the cluster to disable the 'jobs' API as described at:
-http://doc.arvados.org/install/install-api-server.html#disable_api_methods
+The 'jobs' API is no longer supported.
 *******************************""")
+            exit(1)
 
         self.loadingContext = ArvLoadingContext(vars(arvargs))
         self.loadingContext.fetcher_constructor = self.fetcher_constructor
@@ -259,50 +253,37 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
         activity statuses, for example in the RuntimeStatusLoggingHandler.
         """
         with self.workflow_eval_lock:
-            current = arvados_cwl.util.get_current_container(self.api, self.num_retries, logger)
+            current = None
+            try:
+                current = arvados_cwl.util.get_current_container(self.api, self.num_retries, logger)
+            except Exception as e:
+                logger.info("Couldn't get current container: %s", e)
             if current is None:
                 return
             runtime_status = current.get('runtime_status', {})
-            # In case of status being an error, only report the first one.
-            if kind == 'error':
-                if not runtime_status.get('error'):
-                    runtime_status.update({
-                        'error': message
-                    })
-                    if detail is not None:
-                        runtime_status.update({
-                            'errorDetail': detail
-                        })
-                # Further errors are only mentioned as a count.
-                else:
-                    # Get anything before an optional 'and N more' string.
-                    try:
-                        error_msg = re.match(
-                            r'^(.*?)(?=\s*\(and \d+ more\)|$)', runtime_status.get('error')).groups()[0]
-                        more_failures = re.match(
-                            r'.*\(and (\d+) more\)', runtime_status.get('error'))
-                    except TypeError:
-                        # Ignore tests stubbing errors
-                        return
-                    if more_failures:
-                        failure_qty = int(more_failures.groups()[0])
-                        runtime_status.update({
-                            'error': "%s (and %d more)" % (error_msg, failure_qty+1)
-                        })
-                    else:
-                        runtime_status.update({
-                            'error': "%s (and 1 more)" % error_msg
-                        })
-            elif kind in ['warning', 'activity']:
-                # Record the last warning/activity status without regard of
-                # previous occurences.
+            if kind in ('error', 'warning'):
+                updatemessage = runtime_status.get(kind, "")
+                if not updatemessage:
+                    updatemessage = message
+
+                # Subsequent messages tacked on in detail
+                updatedetail = runtime_status.get(kind+'Detail', "")
+                maxlines = 40
+                if updatedetail.count("\n") < maxlines:
+                    if updatedetail:
+                        updatedetail += "\n"
+                    updatedetail += message + "\n"
+
+                    if detail:
+                        updatedetail += detail + "\n"
+
+                    if updatedetail.count("\n") >= maxlines:
+                        updatedetail += "\nSome messages may have been omitted.  Check the full log."
+
                 runtime_status.update({
-                    kind: message
+                    kind: updatemessage,
+                    kind+'Detail': updatedetail,
                 })
-                if detail is not None:
-                    runtime_status.update({
-                        kind+"Detail": detail
-                    })
             else:
                 # Ignore any other status kind
                 return
@@ -339,7 +320,7 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
         return "[%s %s]" % (self.work_api[0:-1], obj.name)
 
     def poll_states(self):
-        """Poll status of jobs or containers listed in the processes dict.
+        """Poll status of containers listed in the processes dict.
 
         Runs in a separate thread.
         """
@@ -360,8 +341,6 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
                 begin_poll = time.time()
                 if self.work_api == "containers":
                     table = self.poll_api.container_requests()
-                elif self.work_api == "jobs":
-                    table = self.poll_api.jobs()
 
                 pageSize = self.poll_api._rootDesc.get('maxItemsPerResponse', 1000)
 
@@ -410,18 +389,11 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
 
     def check_features(self, obj, parentfield=""):
         if isinstance(obj, dict):
-            if obj.get("writable") and self.work_api != "containers":
-                raise SourceLine(obj, "writable", UnsupportedRequirement).makeError("InitialWorkDir feature 'writable: true' not supported with --api=jobs")
             if obj.get("class") == "DockerRequirement":
                 if obj.get("dockerOutputDirectory"):
-                    if self.work_api != "containers":
-                        raise SourceLine(obj, "dockerOutputDirectory", UnsupportedRequirement).makeError(
-                            "Option 'dockerOutputDirectory' of DockerRequirement not supported with --api=jobs.")
                     if not obj.get("dockerOutputDirectory").startswith('/'):
                         raise SourceLine(obj, "dockerOutputDirectory", validate.ValidationException).makeError(
                             "Option 'dockerOutputDirectory' must be an absolute path.")
-            if obj.get("class") == "http://commonwl.org/cwltool#Secrets" and self.work_api != "containers":
-                raise SourceLine(obj, "class", UnsupportedRequirement).makeError("Secrets not supported with --api=jobs")
             if obj.get("class") == "InplaceUpdateRequirement":
                 if obj["inplaceUpdate"] and parentfield == "requirements":
                     raise SourceLine(obj, "class", UnsupportedRequirement).makeError("InplaceUpdateRequirement not supported for keep collections.")
@@ -462,7 +434,7 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
             srccollection = sp[0][5:]
             try:
                 reader = self.collection_cache.get(srccollection)
-                srcpath = "/".join(sp[1:]) if len(sp) > 1 else "."
+                srcpath = urllib.parse.unquote("/".join(sp[1:]) if len(sp) > 1 else ".")
                 final.copy(srcpath, v.target, source_collection=reader, overwrite=False)
             except arvados.errors.ArgumentError as e:
                 logger.error("Creating CollectionReader for '%s' '%s': %s", k, v, e)
@@ -521,14 +493,7 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
                                               }).execute(num_retries=self.num_retries)
             except Exception:
                 logger.exception("Setting container output")
-                return
-        elif self.work_api == "jobs" and "TASK_UUID" in os.environ:
-            self.api.job_tasks().update(uuid=os.environ["TASK_UUID"],
-                                   body={
-                                       'output': self.final_output_collection.portable_data_hash(),
-                                       'success': self.final_status == "success",
-                                       'progress':1.0
-                                   }).execute(num_retries=self.num_retries)
+                raise
 
     def apply_reqs(self, job_order_object, tool):
         if "https://w3id.org/cwl/cwl#requirements" in job_order_object:
@@ -542,10 +507,15 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
             for req in job_reqs:
                 tool.requirements.append(req)
 
-    def arv_executor(self, tool, job_order, runtimeContext, logger=None):
+    def arv_executor(self, updated_tool, job_order, runtimeContext, logger=None):
         self.debug = runtimeContext.debug
 
-        tool.visit(self.check_features)
+        workbench1 = self.api.config()["Services"]["Workbench1"]["ExternalURL"]
+        workbench2 = self.api.config()["Services"]["Workbench2"]["ExternalURL"]
+        controller = self.api.config()["Services"]["Controller"]["ExternalURL"]
+        logger.info("Using cluster %s (%s)", self.api.config()["ClusterID"], workbench2 or workbench1 or controller)
+
+        updated_tool.visit(self.check_features)
 
         self.project_uuid = runtimeContext.project_uuid
         self.pipeline = None
@@ -565,28 +535,41 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
         if runtimeContext.submit_request_uuid and self.work_api != "containers":
             raise Exception("--submit-request-uuid requires containers API, but using '{}' api".format(self.work_api))
 
+        default_storage_classes = ",".join([k for k,v in self.api.config().get("StorageClasses", {"default": {"Default": True}}).items() if v.get("Default") is True])
+        if runtimeContext.storage_classes == "default":
+            runtimeContext.storage_classes = default_storage_classes
+        if runtimeContext.intermediate_storage_classes == "default":
+            runtimeContext.intermediate_storage_classes = default_storage_classes
+
         if not runtimeContext.name:
-            runtimeContext.name = self.name = tool.tool.get("label") or tool.metadata.get("label") or os.path.basename(tool.tool["id"])
+            runtimeContext.name = self.name = updated_tool.tool.get("label") or updated_tool.metadata.get("label") or os.path.basename(updated_tool.tool["id"])
 
         # Upload local file references in the job order.
         job_order = upload_job_order(self, "%s input" % runtimeContext.name,
-                                     tool, job_order)
+                                     updated_tool, job_order)
+
+        # the last clause means: if it is a command line tool, and we
+        # are going to wait for the result, and always_submit_runner
+        # is false, then we don't submit a runner process.
 
         submitting = (runtimeContext.update_workflow or
                       runtimeContext.create_workflow or
                       (runtimeContext.submit and not
-                       (tool.tool["class"] == "CommandLineTool" and
+                       (updated_tool.tool["class"] == "CommandLineTool" and
                         runtimeContext.wait and
                         not runtimeContext.always_submit_runner)))
 
         loadingContext = self.loadingContext.copy()
         loadingContext.do_validate = False
-        loadingContext.do_update = False
         if submitting:
+            loadingContext.do_update = False
             # Document may have been auto-updated. Reload the original
             # document with updating disabled because we want to
-            # submit the original document, not the auto-updated one.
-            tool = load_tool(tool.tool["id"], loadingContext)
+            # submit the document with its original CWL version, not
+            # the auto-updated one.
+            tool = load_tool(updated_tool.tool["id"], loadingContext)
+        else:
+            tool = updated_tool
 
         # Upload direct dependencies of workflow steps, get back mapping of files to keep references.
         # Also uploads docker images.
@@ -604,25 +587,16 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
         existing_uuid = runtimeContext.update_workflow
         if existing_uuid or runtimeContext.create_workflow:
             # Create a pipeline template or workflow record and exit.
-            if self.work_api == "jobs":
-                tmpl = RunnerTemplate(self, tool, job_order,
-                                      runtimeContext.enable_reuse,
-                                      uuid=existing_uuid,
-                                      submit_runner_ram=runtimeContext.submit_runner_ram,
-                                      name=runtimeContext.name,
-                                      merged_map=merged_map,
-                                      loadingContext=loadingContext)
-                tmpl.save()
-                # cwltool.main will write our return value to stdout.
-                return (tmpl.uuid, "success")
-            elif self.work_api == "containers":
-                return (upload_workflow(self, tool, job_order,
+            if self.work_api == "containers":
+                uuid = upload_workflow(self, tool, job_order,
                                         self.project_uuid,
                                         uuid=existing_uuid,
                                         submit_runner_ram=runtimeContext.submit_runner_ram,
                                         name=runtimeContext.name,
-                                        merged_map=merged_map),
-                        "success")
+                                        merged_map=merged_map,
+                                        submit_runner_image=runtimeContext.submit_runner_image)
+                self.stdout.write(uuid + "\n")
+                return (None, "success")
 
         self.apply_reqs(job_order, tool)
 
@@ -641,12 +615,6 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
             runtimeContext.docker_outdir = "/var/spool/cwl"
             runtimeContext.tmpdir = "/tmp"
             runtimeContext.docker_tmpdir = "/tmp"
-        elif self.work_api == "jobs":
-            if runtimeContext.priority != DEFAULT_PRIORITY:
-                raise Exception("--priority not implemented for jobs API.")
-            runtimeContext.outdir = "$(task.outdir)"
-            runtimeContext.docker_outdir = "$(task.outdir)"
-            runtimeContext.tmpdir = "$(task.tmpdir)"
 
         if runtimeContext.priority < 1 or runtimeContext.priority > 1000:
             raise Exception("--priority must be in the range 1..1000.")
@@ -670,40 +638,23 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
         if runtimeContext.submit:
             # Submit a runner job to run the workflow for us.
             if self.work_api == "containers":
-                if tool.tool["class"] == "CommandLineTool" and runtimeContext.wait and (not runtimeContext.always_submit_runner):
-                    runtimeContext.runnerjob = tool.tool["id"]
+                if submitting:
+                    tool = RunnerContainer(self, updated_tool,
+                                           tool, loadingContext, runtimeContext.enable_reuse,
+                                           self.output_name,
+                                           self.output_tags,
+                                           submit_runner_ram=runtimeContext.submit_runner_ram,
+                                           name=runtimeContext.name,
+                                           on_error=runtimeContext.on_error,
+                                           submit_runner_image=runtimeContext.submit_runner_image,
+                                           intermediate_output_ttl=runtimeContext.intermediate_output_ttl,
+                                           merged_map=merged_map,
+                                           priority=runtimeContext.priority,
+                                           secret_store=self.secret_store,
+                                           collection_cache_size=runtimeContext.collection_cache_size,
+                                           collection_cache_is_default=self.should_estimate_cache_size)
                 else:
-                    tool = RunnerContainer(self, tool, loadingContext, runtimeContext.enable_reuse,
-                                                self.output_name,
-                                                self.output_tags,
-                                                submit_runner_ram=runtimeContext.submit_runner_ram,
-                                                name=runtimeContext.name,
-                                                on_error=runtimeContext.on_error,
-                                                submit_runner_image=runtimeContext.submit_runner_image,
-                                                intermediate_output_ttl=runtimeContext.intermediate_output_ttl,
-                                                merged_map=merged_map,
-                                                priority=runtimeContext.priority,
-                                                secret_store=self.secret_store,
-                                                collection_cache_size=runtimeContext.collection_cache_size,
-                                                collection_cache_is_default=self.should_estimate_cache_size)
-            elif self.work_api == "jobs":
-                tool = RunnerJob(self, tool, loadingContext, runtimeContext.enable_reuse,
-                                      self.output_name,
-                                      self.output_tags,
-                                      submit_runner_ram=runtimeContext.submit_runner_ram,
-                                      name=runtimeContext.name,
-                                      on_error=runtimeContext.on_error,
-                                      submit_runner_image=runtimeContext.submit_runner_image,
-                                      merged_map=merged_map)
-        elif runtimeContext.cwl_runner_job is None and self.work_api == "jobs":
-            # Create pipeline for local run
-            self.pipeline = self.api.pipeline_instances().create(
-                body={
-                    "owner_uuid": self.project_uuid,
-                    "name": runtimeContext.name if runtimeContext.name else shortname(tool.tool["id"]),
-                    "components": {},
-                    "state": "RunningOnClient"}).execute(num_retries=self.num_retries)
-            logger.info("Pipeline instance %s", self.pipeline["uuid"])
+                    runtimeContext.runnerjob = tool.tool["id"]
 
         if runtimeContext.cwl_runner_job is not None:
             self.uuid = runtimeContext.cwl_runner_job.get('uuid')
@@ -715,7 +666,8 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
         if runtimeContext.submit and not runtimeContext.wait:
             runnerjob = next(jobiter)
             runnerjob.run(runtimeContext)
-            return (runnerjob.uuid, "success")
+            self.stdout.write(runnerjob.uuid+"\n")
+            return (None, "success")
 
         current_container = arvados_cwl.util.get_current_container(self.api, self.num_retries, logger)
         if current_container:
@@ -805,13 +757,21 @@ http://doc.arvados.org/install/install-api-server.html#disable_api_methods
 
         if runtimeContext.submit and isinstance(tool, Runner):
             logger.info("Final output collection %s", tool.final_output)
+            if workbench2 or workbench1:
+                logger.info("Output at %scollections/%s", workbench2 or workbench1, tool.final_output)
         else:
             if self.output_name is None:
                 self.output_name = "Output of %s" % (shortname(tool.tool["id"]))
             if self.output_tags is None:
                 self.output_tags = ""
 
-            storage_classes = runtimeContext.storage_classes.strip().split(",")
+            storage_classes = ""
+            storage_class_req, _ = tool.get_requirement("http://arvados.org/cwl#OutputStorageClass")
+            if storage_class_req and storage_class_req.get("finalStorageClass"):
+                storage_classes = aslist(storage_class_req["finalStorageClass"])
+            else:
+                storage_classes = runtimeContext.storage_classes.strip().split(",")
+
             self.final_output, self.final_output_collection = self.make_output_collection(self.output_name, storage_classes, self.output_tags, self.final_output)
             self.set_crunch_output()