Turn on more CWL tests

src/toil/batchSystems/kubernetes.py

@@ -463,7 +463,7 @@ def issueBatchJob(self, jobDesc):
 
         # Try the job as local
         localID = self.handleLocalJob(jobDesc)
-        if localID:
+        if localID is not None:
             # It is a local job
             return localID
         else:

src/toil/job.py

@@ -472,6 +472,10 @@ def makeString(x):
         # the logging has been captured to be reported on the leader.
         self.logJobStoreFileID = None
 
+        # Every time we update a job description in place in the job store, we
+        # increment this.
+        self._generation = 0
+
     def serviceHostIDsInBatches(self):
         """
         Get an iterator over all batches of service host job IDs that can be
@@ -635,6 +639,8 @@ def replace(self, other):
         self.filesToDelete += other.filesToDelete
         self.jobsToDelete += other.jobsToDelete
 
+        self._generation = other._generation
+
     def addChild(self, childID):
         """
         Make the job with the given ID a child of the described job.
@@ -814,6 +820,8 @@ def __str__(self):
         if self.jobStoreID is not None:
             printedName += ' ' + str(self.jobStoreID)
 
+        printedName += ' v' + str(self._generation)
+
         return printedName
 
     # Not usable as a key (not hashable) and doesn't have any value-equality.
@@ -823,6 +831,16 @@ def __str__(self):
     def __repr__(self):
         return '%s( **%r )' % (self.__class__.__name__, self.__dict__)
 
+    def pre_update_hook(self) -> None:
+        """
+        Called by the job store before pickling and saving a created or updated
+        version of a job.
+        """
+
+        self._generation += 1
+        logger.debug("New generation: %s", self)
+
+
 
 class ServiceJobDescription(JobDescription):
     """
@@ -1448,7 +1466,8 @@ def registerPromise(self, path):
             raise JobPromiseConstraintError(self)
         # TODO: can we guarantee self.jobStoreID is populated and so pass that here?
         with self._promiseJobStore.writeFileStream() as (fileHandle, jobStoreFileID):
-            promise = UnfulfilledPromiseSentinel(str(self.description), False)
+            promise = UnfulfilledPromiseSentinel(str(self.description), jobStoreFileID, False)
+            logger.debug('Issuing promise %s for result of %s', jobStoreFileID, self.description)
             pickle.dump(promise, fileHandle, pickle.HIGHEST_PROTOCOL)
         self._rvs[path].append(jobStoreFileID)
         return self._promiseJobStore.config.jobStore, jobStoreFileID
@@ -1874,7 +1893,7 @@ def _fulfillPromises(self, returnValues, jobStore):
                 # either case, we just pass it on.
                 promisedValue = returnValues
             else:
-                # If there is an path ...
+                # If there is a path ...
                 if isinstance(returnValues, Promise):
                     # ... and the value itself is a Promise, we need to created a new, narrower
                     # promise and pass it on.
@@ -1888,8 +1907,11 @@ def _fulfillPromises(self, returnValues, jobStore):
                 # File may be gone if the job is a service being re-run and the accessing job is
                 # already complete.
                 if jobStore.fileExists(promiseFileStoreID):
+                    logger.debug("Resolve promise %s from %s with a %s", promiseFileStoreID, self, type(promisedValue))
                     with jobStore.updateFileStream(promiseFileStoreID) as fileHandle:
                         pickle.dump(promisedValue, fileHandle, pickle.HIGHEST_PROTOCOL)
+                else:
+                    logger.debug("Do not resolve promise %s from %s because it is no longer needed", promiseFileStoreID, self)
 
     # Functions associated with Job.checkJobGraphAcyclic to establish that the job graph does not
     # contain any cycles of dependencies:
@@ -2381,7 +2403,6 @@ def _jobName(self):
         """
         return self._description.displayName
 
-
 class JobException(Exception):
     """
     General job exception.
@@ -2577,7 +2598,7 @@ class EncapsulatedJob(Job):
     predecessors automatically. Care should be exercised to ensure the encapsulated job has the
     proper set of predecessors.
 
-    The return value of an encapsulatd job (as accessed by the :func:`toil.job.Job.rv` function)
+    The return value of an encapsulated job (as accessed by the :func:`toil.job.Job.rv` function)
     is the return value of the root job, e.g. A().encapsulate().rv() and A().rv() will resolve to
     the same value after A or A.encapsulate() has been run.
     """
@@ -2930,17 +2951,19 @@ def convertPromises(kwargs):
 class UnfulfilledPromiseSentinel:
     """This should be overwritten by a proper promised value. Throws an
     exception when unpickled."""
-    def __init__(self, fulfillingJobName, unpickled):
+    def __init__(self, fulfillingJobName, file_id: str, unpickled):
         self.fulfillingJobName = fulfillingJobName
+        self.file_id = file_id
 
     @staticmethod
     def __setstate__(stateDict):
         """Only called when unpickling. This won't be unpickled unless the
         promise wasn't resolved, so we throw an exception."""
         jobName = stateDict['fulfillingJobName']
-        raise RuntimeError("This job was passed a promise that wasn't yet resolved when it "
-                           "ran. The job {jobName} that fulfills this promise hasn't yet "
-                           "finished. This means that there aren't enough constraints to "
-                           "ensure the current job always runs after {jobName}. Consider adding a "
-                           "follow-on indirection between this job and its parent, or adding "
-                           "this job as a child/follow-on of {jobName}.".format(jobName=jobName))
+        file_id = stateDict['file_id']
+        raise RuntimeError(f"This job was passed promise {file_id} that wasn't yet resolved when it "
+                           f"ran. The job {jobName} that fulfills this promise hasn't yet "
+                           f"finished. This means that there aren't enough constraints to "
+                           f"ensure the current job always runs after {jobName}. Consider adding a "
+                           f"follow-on indirection between this job and its parent, or adding "
+                           f"this job as a child/follow-on of {jobName}.")

src/toil/jobStores/abstractJobStore.py

@@ -744,6 +744,8 @@ def create(self, jobDescription):
         """
         Writes the given JobDescription to the job store. The job must have an ID assigned already.
 
+        Must call jobDescription.pre_update_hook()
+
         :return: The JobDescription passed.
         :rtype: toil.job.JobDescription
         """
@@ -817,6 +819,8 @@ def update(self, jobDescription):
         """
         Persists changes to the state of the given JobDescription in this store atomically.
 
+        Must call jobDescription.pre_update_hook()
+
         :param toil.job.JobDescription job: the job to write to this job store
         """
         raise NotImplementedError()

src/toil/jobStores/aws/jobStore.py

@@ -306,6 +306,8 @@ def batch(self):
                    range(0, len(self._batchedUpdates), self.jobsPerBatchInsert)]
 
         for batch in batches:
+            for jobDescription in batch:
+                jobDescription.pre_update_hook()
             items = {compat_bytes(jobDescription.jobStoreID): self._awsJobToItem(jobDescription) for jobDescription in batch}
             for attempt in retry_sdb():
                 with attempt:
@@ -359,6 +361,7 @@ def load(self, jobStoreID):
 
     def update(self, jobDescription):
         logger.debug("Updating job %s", jobDescription.jobStoreID)
+        jobDescription.pre_update_hook()
         item = self._awsJobToItem(jobDescription)
         for attempt in retry_sdb():
             with attempt:

src/toil/jobStores/fileJobStore.py

@@ -224,6 +224,8 @@ def update(self, job):
         assert job.jobStoreID is not None, f"Tried to update job {job} without an ID"
         assert not isinstance(job.jobStoreID, TemporaryID), f"Tried to update job {job} without an assigned ID"
 
+        job.pre_update_hook()
+
         # The job is serialised to a file suffixed by ".new"
         # We insist on creating the file; an existing .new file indicates
         # multiple simultaneous attempts to update the job, which will lose

src/toil/jobStores/googleJobStore.py

@@ -181,6 +181,7 @@ def assignID(self, jobDescription):
 
     def create(self, jobDescription):
         # TODO: we don't implement batching, but we probably should.
+        jobDescription.pre_update_hook()
         self._writeBytes(jobDescription.jobStoreID, pickle.dumps(jobDescription, protocol=pickle.HIGHEST_PROTOCOL))
         return jobDescription
 
@@ -212,6 +213,7 @@ def load(self, jobStoreID):
         return job
 
     def update(self, job):
+        job.pre_update_hook()
         self._writeBytes(job.jobStoreID, pickle.dumps(job, protocol=pickle.HIGHEST_PROTOCOL), update=True)
 
     @googleRetry

src/toil/leader.py

@@ -23,15 +23,16 @@
 import pickle
 import sys
 import time
+from typing import List
 
 import enlighten
 
 from toil import resolveEntryPoint
 from toil.batchSystems import DeadlockException
 from toil.batchSystems.abstractBatchSystem import BatchJobExitReason
 from toil.common import Toil, ToilMetrics
-from toil.job import CheckpointJobDescription, ServiceJobDescription
-from toil.jobStores.abstractJobStore import NoSuchJobException
+from toil.job import JobDescription, CheckpointJobDescription, ServiceJobDescription
+from toil.jobStores.abstractJobStore import AbstractJobStore, NoSuchJobException
 from toil.lib.conversions import bytes2human
 from toil.lib.throttle import LocalThrottle
 from toil.provisioners.abstractProvisioner import AbstractProvisioner
@@ -41,10 +42,11 @@
 from toil.toilState import ToilState
 
 try:
-    from toil.cwl.cwltoil import CWL_INTERNAL_JOBS
+    from toil.cwl.cwltoil import CWL_INTERNAL_JOBS, CWL_UNSUPPORTED_REQUIREMENT_EXIT_CODE
 except ImportError:
     # CWL extra not installed
     CWL_INTERNAL_JOBS = ()
+    CWL_UNSUPPORTED_REQUIREMENT_EXIT_CODE = 0
 
 
 logger = logging.getLogger( __name__ )
@@ -66,8 +68,18 @@
 
 
 class FailedJobsException(Exception):
-    def __init__(self, jobStoreLocator, failedJobs, jobStore):
+    def __init__(self, jobStoreLocator: str, failedJobs: List[JobDescription], jobStore: AbstractJobStore, exit_code: int = 1):
+        """
+        Make an exception to report failed jobs.
+
+        :param jobStoreLocator: The job store locator that says what job store has the failed jobs.
+        :param failedJobs: All the failed jobs
+        :param jobStore: The actual open job store with the failed jobs in it.
+        :param exit_code: Recommended process exit code.
+
+        """
         self.msg = "The job store '%s' contains %i failed jobs" % (jobStoreLocator, len(failedJobs))
+        self.exit_code = exit_code
         try:
             self.msg += ": %s" % ", ".join((str(failedJob) for failedJob in failedJobs))
             for jobDesc in failedJobs:
@@ -191,6 +203,11 @@ def __init__(self, config, batchSystem, provisioner: AbstractProvisioner, jobSto
         self.PROGRESS_BAR_FORMAT = ('{desc}{desc_pad}{percentage:3.0f}%|{bar}| {count:{len_total}d}/{total:d} '
                                     '({count_1:d} failures) [{elapsed}<{eta}, {rate:.2f}{unit_pad}{unit}/s]')
 
+        # What exit code should the process use if the workflow failed?
+        # Needed in case a worker detects a CWL issue that a CWL runner must
+        # report to its caller.
+        self.recommended_fail_exit_code = 1
+
         # TODO: No way to set background color on the terminal for the bar.
 
     def run(self):
@@ -258,7 +275,8 @@ def run(self):
 
             if len(self.toilState.totalFailedJobs):
                 logger.info("Failed jobs at end of the run: %s", ' '.join(str(job) for job in self.toilState.totalFailedJobs))
-                raise FailedJobsException(self.config.jobStore, self.toilState.totalFailedJobs, self.jobStore)
+                raise FailedJobsException(self.config.jobStore, self.toilState.totalFailedJobs,
+                                          self.jobStore, exit_code=self.recommended_fail_exit_code)
 
             return self.jobStore.getRootJobReturnValue()
 
@@ -568,6 +586,13 @@ def _gatherUpdatedJobs(self, updatedJobTuple):
             else:
                 logger.warning(f'Job failed with exit value {exitStatus}: {updatedJob}\n'
                                f'Exit reason: {exitReason}')
+                if exitStatus == CWL_UNSUPPORTED_REQUIREMENT_EXIT_CODE:
+                    # This is a CWL job informing us that the workflow is
+                    # asking things of us that Toil can't do. When we raise an
+                    # exception because of this, make sure to forward along
+                    # this exit code.
+                    logger.warning("This indicates an unsupported CWL requirement!")
+                    self.recommended_fail_exit_code = CWL_UNSUPPORTED_REQUIREMENT_EXIT_CODE
             if self.toilMetrics:
                 self.toilMetrics.logCompletedJob(updatedJob)
             self.processFinishedJob(jobID, exitStatus, wallTime=wallTime, exitReason=exitReason)

src/toil/test/__init__.py

@@ -340,11 +340,16 @@ def needs_kubernetes(test_item):
     test_item = _mark_test('kubernetes', test_item)
     try:
         import kubernetes
-        kubernetes.config.load_kube_config()
+        try:
+            kubernetes.config.load_kube_config()
+        except kubernetes.config.ConfigException:
+            try:
+                kubernetes.config.load_incluster_config()
+            except kubernetes.config.ConfigException:
+                return unittest.skip("Configure Kubernetes (~/.kube/config, $KUBECONFIG, "
+                                     "or current pod) to include this test.")(test_item)
     except ImportError:
         return unittest.skip("Install Toil with the 'kubernetes' extra to include this test.")(test_item)
-    except TypeError:
-        return unittest.skip("Configure Kubernetes (~/.kube/config) to include this test.")(test_item)
     return test_item
 
 

src/toil/test/cwl/cwlTest.py

@@ -504,9 +504,10 @@ def run_kubernetes_cwl_conformance(self, **kwargs):
                                          **kwargs)
     @slow
     @needs_kubernetes
-    def test_kubernetes_cwl_20(self):
+    @pytest.mark.timeout(CONFORMANCE_TEST_TIMEOUT)
+    def test_kubernetes_cwl_group(self):
         for caching in [True, False]:
-            self.run_kubernetes_cwl_conformance(selected_tests="20", caching=caching)
+            self.run_kubernetes_cwl_conformance(selected_tests="20,35,39,42,56", caching=caching)
 
 @needs_cwl
 class CWLSmallTests(ToilTest):

src/toil/worker.py

@@ -22,9 +22,11 @@
 import shutil
 import signal
 import socket
+import stat
 import sys
 import time
 import traceback
+from typing import Callable, Any
 from contextlib import contextmanager
 
 from toil import logProcessContext
@@ -39,10 +41,14 @@
 from toil.statsAndLogging import configure_root_logger, set_log_level
 
 try:
-    from toil.cwl.cwltoil import CWL_INTERNAL_JOBS
+    from toil.cwl.cwltoil import (CWL_INTERNAL_JOBS,
+                                  CWL_UNSUPPORTED_REQUIREMENT_EXIT_CODE,
+                                  CWL_UNSUPPORTED_REQUIREMENT_EXCEPTION)
 except ImportError:
     # CWL extra not installed
     CWL_INTERNAL_JOBS = ()
+    CWL_UNSUPPORTED_REQUIREMENT_EXIT_CODE = None
+    CWL_UNSUPPORTED_REQUIREMENT_EXCEPTION = type(None)
 
 logger = logging.getLogger(__name__)
 
@@ -275,6 +281,7 @@ def workerScript(jobStore, config, jobName, jobStoreID, redirectOutputToLogFile=
     ##########################################
 
     jobAttemptFailed = False
+    failure_exit_code = 1
     statsDict = MagicExpando()
     statsDict.jobs = []
     statsDict.workers.logsToMaster = []
@@ -492,9 +499,13 @@ def workerScript(jobStore, config, jobName, jobStoreID, redirectOutputToLogFile=
     ##########################################
     #Trapping where worker goes wrong
     ##########################################
-    except: #Case that something goes wrong in worker
+    except Exception as e: #Case that something goes wrong in worker
         traceback.print_exc()
         logger.error("Exiting the worker because of a failed job on host %s", socket.gethostname())
+        if isinstance(e, CWL_UNSUPPORTED_REQUIREMENT_EXCEPTION):
+            # We need to inform the leader that this is a CWL workflow problem
+            # and it needs to inform its caller.
+            failure_exit_code = CWL_UNSUPPORTED_REQUIREMENT_EXIT_CODE
         AbstractFileStore._terminateEvent.set()
 
     ##########################################
@@ -587,7 +598,17 @@ def workerScript(jobStore, config, jobName, jobStoreID, redirectOutputToLogFile=
     #Remove the temp dir
     cleanUp = config.cleanWorkDir
     if cleanUp == 'always' or (cleanUp == 'onSuccess' and not jobAttemptFailed) or (cleanUp == 'onError' and jobAttemptFailed):
-        shutil.rmtree(localWorkerTempDir)
+        def make_parent_writable(func: Callable[[str], Any], path: str, _) -> None:
+            """
+            When encountering an error removing a file or directory, make sure
+            the parent directory is writable.
+
+            cwltool likes to lock down directory permissions, and doesn't clean
+            up after itself.
+            """
+            # Just chmod it for rwx for user. This can't work anyway if it isn't ours.
+            os.chmod(os.path.dirname(path),  stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+        shutil.rmtree(localWorkerTempDir, onerror=make_parent_writable)
 
     #This must happen after the log file is done with, else there is no place to put the log
     if (not jobAttemptFailed) and jobDesc.command == None and next(jobDesc.successorsAndServiceHosts(), None) is None:
@@ -597,7 +618,7 @@ def workerScript(jobStore, config, jobName, jobStoreID, redirectOutputToLogFile=
         jobStore.delete(jobDesc.jobStoreID)
 
     if jobAttemptFailed:
-        return 1
+        return failure_exit_code
     else:
         return 0