Running a basic CWL workflow¶

The Common Workflow Language (CWL) is an emerging standard for writing workflows that are portable across multiple workflow engines and platforms. Running CWL workflows using Toil is easy.

cwlVersion: v1.0
class: CommandLineTool
baseCommand: echo
stdout: output.txt
inputs:


  message:


    type: string


    inputBinding:


      position: 1
outputs:


  output:


    type: stdout

message: Hello world!

$ toil-cwl-runner example.cwl example-job.yaml

$ cat output.txt
Hello world!

Congratulations! You've run your first Toil workflow using the default Batch System, single_machine, and the default file job store (which was placed in a temporary directory for you by toil-cwl-runner).

Toil uses batch systems to manage the jobs it creates.

The single_machine batch system is primarily used to prepare and debug workflows on a local machine. Once validated, try running them on a full-fledged batch system (see Batch System API). Toil supports many different batch systems such as Kubernetes and Grid Engine; its versatility makes it easy to run your workflow in all kinds of places.

Toil's CWL runner is totally customizable! Run toil-cwl-runner --help to see a complete list of available options.

To learn more about CWL, see the CWL User Guide (from where this example was shamelessly borrowed). For information on using CWL with Toil see the section CWL in Toil. And for an example of CWL on an AWS cluster, have a look at Running a CWL Workflow on AWS.

Running a basic WDL workflow¶

The Workflow Description Language (WDL) is another emerging language for writing workflows that are portable across multiple workflow engines and platforms. Running WDL workflows using Toil is still in alpha, and currently experimental. Toil currently supports basic workflow syntax (see WDL in Toil for more details and examples). Here we go over running a basic WDL helloworld workflow.

workflow write_simple_file {


  call write_file
}
task write_file {


  String message


  command { echo ${message} > wdl-helloworld-output.txt }


  output { File test = "wdl-helloworld-output.txt" }
}

{


  "write_simple_file.write_file.message": "Hello world!"
}

$ toil-wdl-runner wdl-helloworld.wdl wdl-helloworld.json

$ cat wdl-helloworld-output.txt
Hello world!

This will, like the CWL example above, use the single_machine batch system and an automatically-located file job store by default. You can customize Toil's execution of the workflow with command-line options; run toil-wdl-runner --help to learn about them.

To learn more about WDL in general, see the Terra WDL documentation . For more on using WDL in Toil, see WDL in Toil.

Running a basic Python workflow¶

In addition to workflow languages like CWL and WDL, Toil supports running workflows written against its Python API.

An example Toil Python workflow can be run with just three steps:

from toil.common import Toil
from toil.job import Job
def helloWorld(message, memory="1G", cores=1, disk="1G"):


    return f"Hello, world!, here's a message: {message}"
if __name__ == "__main__":


    parser = Job.Runner.getDefaultArgumentParser()


    options = parser.parse_args()


    options.clean = "always"


    with Toil(options) as toil:


        output = toil.start(Job.wrapFn(helloWorld, "You did it!"))


    print(output)

$ python3 helloWorld.py file:my-job-store

For something beyond a "Hello, world!" example, refer to A (more) real-world example.

Toil's customization options are available in Python workflows. Run python3 helloWorld.py --help to see a complete list of available options.

A (more) real-world example¶

For a more detailed example and explanation, we've developed a sample pipeline that merge-sorts a temporary file. This is not supposed to be an efficient sorting program, rather a more fully worked example of what Toil is capable of.

Running the example¶

$ python3 sort.py file:jobStore

NOTE:

$ python3 sort.py file:jobStore \


             --numLines=5000 \


             --lineLength=10 \


             --overwriteOutput=True \


             --workDir=/tmp/

Describing the source code¶

To understand the details of what's going on inside. Let's start with the main() function. It looks like a lot of code, but don't worry---we'll break it down piece by piece.

def main(options=None):


    if not options:


        # deal with command line arguments


        parser = ArgumentParser()


        Job.Runner.addToilOptions(parser)


        parser.add_argument('--numLines', default=defaultLines, help='Number of lines in file to sort.', type=int)


        parser.add_argument('--lineLength', default=defaultLineLen, help='Length of lines in file to sort.', type=int)


        parser.add_argument("--fileToSort", help="The file you wish to sort")


        parser.add_argument("--outputFile", help="Where the sorted output will go")


        parser.add_argument("--overwriteOutput", help="Write over the output file if it already exists.", default=True)


        parser.add_argument("--N", dest="N",


                            help="The threshold below which a serial sort function is used to sort file. "


                                 "All lines must of length less than or equal to N or program will fail",


                            default=10000)


        parser.add_argument('--downCheckpoints', action='store_true',


                            help='If this option is set, the workflow will make checkpoints on its way through'


                                 'the recursive "down" part of the sort')


        parser.add_argument("--sortMemory", dest="sortMemory",


                        help="Memory for jobs that sort chunks of the file.",


                        default=None)


        parser.add_argument("--mergeMemory", dest="mergeMemory",


                        help="Memory for jobs that collate results.",


                        default=None)


        options = parser.parse_args()


    if not hasattr(options, "sortMemory") or not options.sortMemory:


        options.sortMemory = sortMemory


    if not hasattr(options, "mergeMemory") or not options.mergeMemory:


        options.mergeMemory = sortMemory


    # do some input verification


    sortedFileName = options.outputFile or "sortedFile.txt"


    if not options.overwriteOutput and os.path.exists(sortedFileName):


        print(f'Output file {sortedFileName} already exists.  '


              f'Delete it to run the sort example again or use --overwriteOutput=True')


        exit()


    fileName = options.fileToSort


    if options.fileToSort is None:


        # make the file ourselves


        fileName = 'fileToSort.txt'


        if os.path.exists(fileName):


            print(f'Sorting existing file: {fileName}')


        else:


            print(f'No sort file specified. Generating one automatically called: {fileName}.')


            makeFileToSort(fileName=fileName, lines=options.numLines, lineLen=options.lineLength)


    else:


        if not os.path.exists(options.fileToSort):


            raise RuntimeError("File to sort does not exist: %s" % options.fileToSort)


    if int(options.N) <= 0:


        raise RuntimeError("Invalid value of N: %s" % options.N)


    # Now we are ready to run


    with Toil(options) as workflow:


        sortedFileURL = 'file://' + os.path.abspath(sortedFileName)


        if not workflow.options.restart:


            sortFileURL = 'file://' + os.path.abspath(fileName)


            sortFileID = workflow.importFile(sortFileURL)


            sortedFileID = workflow.start(Job.wrapJobFn(setup,


                                                        sortFileID,


                                                        int(options.N),


                                                        options.downCheckpoints,


                                                        options=options,


                                                        memory=sortMemory))


        else:


            sortedFileID = workflow.restart()


        workflow.exportFile(sortedFileID, sortedFileURL)

First we make a parser to process command line arguments using the argparse module. It's important that we add the call to Job.Runner.addToilOptions() to initialize our parser with all of Toil's default options. Then we add the command line arguments unique to this workflow, and parse the input. The help message listed with the arguments should give you a pretty good idea of what they can do.

Next we do a little bit of verification of the input arguments. The option --fileToSort allows you to specify a file that needs to be sorted. If this option isn't given, it's here that we make our own file with the call to makeFileToSort().

Finally we come to the context manager that initializes the workflow. We create a path to the input file prepended with 'file://' as per the documentation for toil.common.Toil() when staging a file that is stored locally. Notice that we have to check whether or not the workflow is restarting so that we don't import the file more than once. Finally we can kick off the workflow by calling toil.common.Toil.start() on the job setup. When the workflow ends we capture its output (the sorted file's fileID) and use that in toil.common.Toil.exportFile() to move the sorted file from the job store back into "userland".

Next let's look at the job that begins the actual workflow, setup.

def setup(job, inputFile, N, downCheckpoints, options):


    """


    Sets up the sort.


    Returns the FileID of the sorted file


    """


    RealtimeLogger.info("Starting the merge sort")


    return job.addChildJobFn(down,


                             inputFile, N, 'root',


                             downCheckpoints,


                             options = options,


                             preemptible=True,


                             memory=sortMemory).rv()

setup really only does two things. First it writes to the logs using Job.log() and then calls addChildJobFn(). Child jobs run directly after the current job. This function turns the 'job function' down into an actual job and passes in the inputs including an optional resource requirement, memory. The job doesn't actually get run until the call to Job.rv(). Once the job down finishes, its output is returned here.

Now we can look at what down does.

def down(job, inputFileStoreID, N, path, downCheckpoints, options, memory=sortMemory):


    """


    Input is a file, a subdivision size N, and a path in the hierarchy of jobs.


    If the range is larger than a threshold N the range is divided recursively and


    a follow on job is then created which merges back the results else


    the file is sorted and placed in the output.


    """


    RealtimeLogger.info("Down job starting: %s" % path)


    # Read the file


    inputFile = job.fileStore.readGlobalFile(inputFileStoreID, cache=False)


    length = os.path.getsize(inputFile)


    if length > N:


        # We will subdivide the file


        RealtimeLogger.critical("Splitting file: %s of size: %s"


                % (inputFileStoreID, length))


        # Split the file into two copies


        midPoint = getMidPoint(inputFile, 0, length)


        t1 = job.fileStore.getLocalTempFile()


        with open(t1, 'w') as fH:


            fH.write(copySubRangeOfFile(inputFile, 0, midPoint+1))


        t2 = job.fileStore.getLocalTempFile()


        with open(t2, 'w') as fH:


            fH.write(copySubRangeOfFile(inputFile, midPoint+1, length))


        # Call down recursively. By giving the rv() of the two jobs as inputs to the follow-on job, up,


        # we communicate the dependency without hindering concurrency.


        result = job.addFollowOnJobFn(up,


                                    job.addChildJobFn(down, job.fileStore.writeGlobalFile(t1), N, path + '/0',


                                                      downCheckpoints, checkpoint=downCheckpoints, options=options,


                                                      preemptible=True, memory=options.sortMemory).rv(),


                                    job.addChildJobFn(down, job.fileStore.writeGlobalFile(t2), N, path + '/1',


                                                      downCheckpoints, checkpoint=downCheckpoints, options=options,


                                                      preemptible=True, memory=options.mergeMemory).rv(),


                                    path + '/up', preemptible=True, options=options, memory=options.sortMemory).rv()


    else:


        # We can sort this bit of the file


        RealtimeLogger.critical("Sorting file: %s of size: %s"


                % (inputFileStoreID, length))


        # Sort the copy and write back to the fileStore


        shutil.copyfile(inputFile, inputFile + '.sort')


        sort(inputFile + '.sort')


        result = job.fileStore.writeGlobalFile(inputFile + '.sort')


    RealtimeLogger.info("Down job finished: %s" % path)


    return result

Down is the recursive part of the workflow. First we read the file into the local filestore by calling job.fileStore.readGlobalFile(). This puts a copy of the file in the temp directory for this particular job. This storage will disappear once this job ends. For a detailed explanation of the filestore, job store, and their interfaces have a look at Managing files within a workflow.

Next down checks the base case of the recursion: is the length of the input file less than N (remember N was an option we added to the workflow in main)? In the base case, we just sort the file, and return the file ID of this new sorted file.

If the base case fails, then the file is split into two new tempFiles using job.fileStore.getLocalTempFile() and the helper function copySubRangeOfFile. Finally we add a follow on Job up with job.addFollowOnJobFn(). We've already seen child jobs. A follow-on Job is a job that runs after the current job and all of its children (and their children and follow-ons) have completed. Using a follow-on makes sense because up is responsible for merging the files together and we don't want to merge the files together until we know they are sorted. Again, the return value of the follow-on job is requested using Job.rv().

Looking at up

def up(job, inputFileID1, inputFileID2, path, options, memory=sortMemory):


    """


    Merges the two files and places them in the output.


    """


    RealtimeLogger.info("Up job starting: %s" % path)


    with job.fileStore.writeGlobalFileStream() as (fileHandle, outputFileStoreID):


        fileHandle = codecs.getwriter('utf-8')(fileHandle)


        with job.fileStore.readGlobalFileStream(inputFileID1) as inputFileHandle1:


            inputFileHandle1 = codecs.getreader('utf-8')(inputFileHandle1)


            with job.fileStore.readGlobalFileStream(inputFileID2) as inputFileHandle2:


                inputFileHandle2 = codecs.getreader('utf-8')(inputFileHandle2)


                RealtimeLogger.info("Merging %s and %s to %s"


                    % (inputFileID1, inputFileID2, outputFileStoreID))


                merge(inputFileHandle1, inputFileHandle2, fileHandle)


        # Cleanup up the input files - these deletes will occur after the completion is successful.


        job.fileStore.deleteGlobalFile(inputFileID1)


        job.fileStore.deleteGlobalFile(inputFileID2)


        RealtimeLogger.info("Up job finished: %s" % path)


        return outputFileStoreID

we see that the two input files are merged together and the output is written to a new file using job.fileStore.writeGlobalFileStream(). After a little cleanup, the output file is returned.

Once the final up finishes and all of the rv() promises are fulfilled, main receives the sorted file's ID which it uses in exportFile to send it to the user.

There are other things in this example that we didn't go over such as Checkpoints and the details of much of the Toil Class API.

At the end of the script the lines

if __name__ == '__main__'


    main()

are included to ensure that the main function is only run once in the '__main__' process invoked by you, the user. In Toil terms, by invoking the script you created the leader process in which the main() function is run. A worker process is a separate process whose sole purpose is to host the execution of one or more jobs defined in that script. In any Toil workflow there is always one leader process, and potentially many worker processes.

When using the single-machine batch system (the default), the worker processes will be running on the same machine as the leader process. With full-fledged batch systems like Kubernetes the worker processes will typically be started on separate machines. The boilerplate ensures that the pipeline is only started once---on the leader---but not when its job functions are imported and executed on the individual workers.

Typing python3 sort.py --help will show the complete list of arguments for the workflow which includes both Toil's and ones defined inside sort.py. A complete explanation of Toil's arguments can be found in Commandline Options.

Logging¶

By default, Toil logs a lot of information related to the current environment in addition to messages from the batch system and jobs. This can be configured with the --logLevel flag. For example, to only log CRITICAL level messages to the screen:

$ python3 sort.py file:jobStore \


             --logLevel=critical \


             --overwriteOutput=True

This hides most of the information we get from the Toil run. For more detail, we can run the pipeline with --logLevel=debug to see a comprehensive output. For more information, see Commandline Options.

Error Handling and Resuming Pipelines¶

With Toil, you can recover gracefully from a bug in your pipeline without losing any progress from successfully completed jobs. To demonstrate this, let's add a bug to our example code to see how Toil handles a failure and how we can resume a pipeline after that happens. Add a bad assertion at line 52 of the example (the first line of down()):

def down(job, inputFileStoreID, N, downCheckpoints, memory=sortMemory):


    ...


    assert 1 == 2, "Test error!"

When we run the pipeline, Toil will show a detailed failure log with a traceback:

$ python3 sort.py file:jobStore
...
---TOIL WORKER OUTPUT LOG---
...
m/j/jobonrSMP    Traceback (most recent call last):
m/j/jobonrSMP      File "toil/src/toil/worker.py", line 340, in main
m/j/jobonrSMP        job._runner(jobGraph=jobGraph, jobStore=jobStore, fileStore=fileStore)
m/j/jobonrSMP      File "toil/src/toil/job.py", line 1270, in _runner
m/j/jobonrSMP        returnValues = self._run(jobGraph, fileStore)
m/j/jobonrSMP      File "toil/src/toil/job.py", line 1217, in _run
m/j/jobonrSMP        return self.run(fileStore)
m/j/jobonrSMP      File "toil/src/toil/job.py", line 1383, in run
m/j/jobonrSMP        rValue = userFunction(*((self,) + tuple(self._args)), **self._kwargs)
m/j/jobonrSMP      File "toil/example.py", line 30, in down
m/j/jobonrSMP        assert 1 == 2, "Test error!"
m/j/jobonrSMP    AssertionError: Test error!

If we try and run the pipeline again, Toil will give us an error message saying that a job store of the same name already exists. By default, in the event of a failure, the job store is preserved so that the workflow can be restarted, starting from the previously failed jobs. We can restart the pipeline by running

$ python3 sort.py file:jobStore \


             --restart \


             --overwriteOutput=True

We can also change the number of times Toil will attempt to retry a failed job:

$ python3 sort.py file:jobStore \


             --retryCount 2 \


             --restart \


             --overwriteOutput=True

You'll now see Toil attempt to rerun the failed job until it runs out of tries. --retryCount is useful for non-systemic errors, like downloading a file that may experience a sporadic interruption, or some other non-deterministic failure.

To successfully restart our pipeline, we can edit our script to comment out line 30, or remove it, and then run

$ python3 sort.py file:jobStore \


             --restart \


             --overwriteOutput=True

The pipeline will run successfully, and the job store will be removed on the pipeline's completion.

Collecting Statistics¶

Please see the Status Command section for more on gathering runtime and resource info on jobs.

Launching a Toil Workflow in AWS¶

After having installed the aws extra for Toil during the Installation and set up AWS (see Preparing your AWS environment), the user can run the basic helloWorld.py script (Running a basic Python workflow) on a VM in AWS just by modifying the run command.

Note that when running in AWS, users can either run the workflow on a single instance or run it on a cluster (which is running across multiple containers on multiple AWS instances). For more information on running Toil workflows on a cluster, see Running in AWS.

Also! Remember to use the Destroy-Cluster Command command when finished to destroy the cluster! Otherwise things may not be cleaned up properly.

$ toil launch-cluster <cluster-name> \


             --clusterType kubernetes \


             --keyPairName <AWS-key-pair-name> \


             --leaderNodeType t2.medium \


             --nodeTypes t2.medium -w 1 \


             --zone us-west-2a

$ toil rsync-cluster --zone us-west-2a <cluster-name> helloWorld.py :/tmp

$ toil ssh-cluster --zone us-west-2a <cluster-name>

$ python3 /tmp/helloWorld.py aws:us-west-2:my-S3-bucket

$ exit

$ toil destroy-cluster --zone us-west-2a <cluster-name>

Running a CWL Workflow on AWS¶

After having installed the aws and cwl extras for Toil during the Installation and set up AWS (see Preparing your AWS environment), the user can run a CWL workflow with Toil on AWS.

Also! Remember to use the Destroy-Cluster Command command when finished to destroy the cluster! Otherwise things may not be cleaned up properly.

$ toil launch-cluster <cluster-name> \


             --clusterType kubernetes \


             --keyPairName <AWS-key-pair-name> \


             --leaderNodeType t2.medium \


             --nodeTypes t2.medium -w 1 \


             --zone us-west-2a

toil rsync-cluster --zone us-west-2a <cluster-name> example.cwl :/tmp
toil rsync-cluster --zone us-west-2a <cluster-name> example-job.yaml :/tmp

$ toil ssh-cluster --zone us-west-2a <cluster-name>

sudo apt-get update
sudo apt-get -y upgrade
sudo apt-get -y dist-upgrade
sudo apt-get -y install git

virtualenv --system-site-packages venv
source venv/bin/activate

(venv) $ toil-cwl-runner \


             --provisioner aws \


             --batchSystem kubernetes \


             --jobStore aws:us-west-2:any-name \


             /tmp/example.cwl /tmp/example-job.yaml

$ toil destroy-cluster --zone us-west-2a <cluster-name>

Running a Workflow with Autoscaling - Cactus¶

Cactus is a reference-free, whole-genome multiple alignment program that can be run on any of the cloud platforms Toil supports.

NOTE:

Also! Remember to use the Destroy-Cluster Command command when finished to destroy the cluster! Otherwise things may not be cleaned up properly.

$ toil launch-cluster <cluster-name> \


             --provisioner <aws, gce> \


             --keyPairName <key-pair-name> \


             --leaderNodeType <type> \


             --nodeType <type> \


             -w 1-2 \


             --zone <zone>

$ export TOIL_AWS_ZONE=us-west-2c

$ toil ssh-cluster --provisioner <aws, gce> <cluster-name>
$ mkdir /root/cact_ex
$ exit

$ toil rsync-cluster --provisioner <aws, gce> <cluster-name> pestis-short-aws-seqFile.txt :/root/cact_ex
$ toil rsync-cluster --provisioner <aws, gce> <cluster-name> GCF_000169655.1_ASM16965v1_genomic.fna :/root/cact_ex
$ toil rsync-cluster --provisioner <aws, gce> <cluster-name> GCF_000006645.1_ASM664v1_genomic.fna :/root/cact_ex
$ toil rsync-cluster --provisioner <aws, gce> <cluster-name> GCF_000182485.1_ASM18248v1_genomic.fna :/root/cact_ex
$ toil rsync-cluster --provisioner <aws, gce> <cluster-name> GCF_000013805.1_ASM1380v1_genomic.fna :/root/cact_ex
$ toil rsync-cluster --provisioner <aws, gce> <cluster-name> setup_leaderNode.sh :/root/cact_ex
$ toil rsync-cluster --provisioner <aws, gce> <cluster-name> blockTrim1.xml :/root/cact_ex
$ toil rsync-cluster --provisioner <aws, gce> <cluster-name> blockTrim3.xml :/root/cact_ex

$ toil ssh-cluster --provisioner <aws, gce> <cluster-name>

$ bash /root/cact_ex/setup_leaderNode.sh
$ source cact_venv/bin/activate
(cact_venv) $ cd cactus
(cact_venv) $ pip install --upgrade .

(cact_venv) $ cactus \


                  --retry 10 \


                  --batchSystem kubernetes \


                  --logDebug \


                  --logFile /logFile_pestis3 \


                  --configFile \


                  /root/cact_ex/blockTrim3.xml <aws, google>:<zone>:cactus-pestis \


                  /root/cact_ex/pestis-short-aws-seqFile.txt \


                  /root/cact_ex/pestis_output3.hal

(cact_venv) $ exit

(venv) $ toil rsync-cluster \


             --provisioner <aws, gce> <cluster-name> \


             :/root/cact_ex/pestis_output3.hal \


             <path-of-folder-on-local-machine>

(venv) $ toil destroy-cluster --provisioner <aws, gce> <cluster-name>

Running CWL Locally¶

To run in local batch mode, provide the CWL file and the input object file:

$ toil-cwl-runner example.cwl example-job.yml

For a simple example of CWL with Toil see Running a basic CWL workflow.

Note for macOS + Docker + Toil¶

When invoking CWL documents that make use of Docker containers if you see errors that look like

docker: Error response from daemon: Mounts denied:
The paths /var/...tmp are not shared from OS X and are not known to Docker.

you may need to add

export TMPDIR=/tmp/docker_tmp

either in your startup file (.bashrc) or add it manually in your shell before invoking toil.

Detailed Usage Instructions¶

Help information can be found by using this toil command:

$ toil-cwl-runner -h

A more detailed example shows how we can specify both Toil and cwltool arguments for our workflow:

$ toil-cwl-runner \


    --singularity \


    --jobStore my_jobStore \


    --batchSystem lsf \


    --workDir `pwd` \


    --outdir `pwd` \


    --logFile cwltoil.log \


    --writeLogs `pwd` \


    --logLevel DEBUG \


    --retryCount 2 \


    --maxLogFileSize 20000000000 \


    --stats \


    standard_bam_processing.cwl \


    inputs.yaml

In this example, we set the following options, which are all passed to Toil:

--singularity: Specifies that all jobs with Docker format containers specified should be run using the Singularity container engine instead of the Docker container engine.

--jobStore: Path to a folder which doesn't exist yet, which will contain the Toil jobstore and all related job-tracking information.

--batchSystem: Use the specified HPC or Cloud-based cluster platform.

--workDir: The directory where all temporary files will be created for the workflow. A subdirectory of this will be set as the $TMPDIR environment variable and this subdirectory can be referenced using the CWL parameter reference $(runtime.tmpdir) in CWL tools and workflows.

--outdir: Directory where final File and Directory outputs will be written. References to these and other output types will be in the JSON object printed to the stdout stream after workflow execution.

--logFile: Path to the main logfile.

--writeLogs: Directory where job logs will be stored. At DEBUG log level, this will contain logs for each Toil job run, as well as stdout/stderr logs for each CWL CommandLineTool that didn't use the stdout/stderr directives to redirect output.

--retryCount: How many times to retry each Toil job.

--maxLogFileSize: Logs that get larger than this value will be truncated.

--stats: Save resources usages in json files that can be collected with the toil stats command after the workflow is done.

--disable-streaming: Does not allow streaming of input files. This is enabled by default for files marked with streamable flag True and only for remote files when the jobStore is not on local machine.

Running CWL in the Cloud¶

To run in cloud and HPC configurations, you may need to provide additional command line parameters to select and configure the batch system to use.

To run a CWL workflow in AWS with toil see Running a CWL Workflow on AWS.

Running CWL within Toil Scripts¶

A CWL workflow can be run from a Toil Python workflow. However, this is not the standard way to run CWL workflows with Toil and doing so comes at the cost of job efficiency. For some use cases, such as running one process on multiple files, it may be useful. For example, if you want to run a CWL workflow with 3 different input files specifying different samples inputs, it could look something like:

import os
import subprocess
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
def initialize_jobs(job):


    job.fileStore.log_to_leader('initialize_jobs')
def runQC(job, cwl_file, cwl_filename, yml_file, yml_filename, outputs_dir, output_num):


    job.fileStore.log_to_leader("runQC")


    tempDir = job.fileStore.getLocalTempDir()


    cwl = job.fileStore.readGlobalFile(cwl_file, userPath=os.path.join(tempDir, cwl_filename))


    yml = job.fileStore.readGlobalFile(yml_file, userPath=os.path.join(tempDir, yml_filename))


    subprocess.check_call(["toil-cwl-runner", cwl, yml])


    output_filename = "output.txt"


    output_file = job.fileStore.writeGlobalFile(output_filename)


    job.fileStore.readGlobalFile(output_file, userPath=os.path.join(outputs_dir, "sample_" + output_num + "_" + output_filename))


    return output_file
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_cwlexample")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        # specify the folder where the cwl and yml files live


        inputs_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cwlExampleFiles")


        # specify where you wish the outputs to be written


        outputs_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cwlExampleFiles")


        job0 = Job.wrapJobFn(initialize_jobs)


        cwl_filename = "hello.cwl"


        cwl_file = toil.importFile("file://" + os.path.abspath(os.path.join(inputs_dir, cwl_filename)))


        # add list of yml config inputs here or import and construct from file


        yml_files = ["hello1.yml", "hello2.yml", "hello3.yml"]


        i = 0


        for yml in yml_files:


            i = i + 1


            yml_file = toil.importFile("file://" + os.path.abspath(os.path.join(inputs_dir, yml)))


            yml_filename = yml


            job = Job.wrapJobFn(runQC, cwl_file, cwl_filename, yml_file, yml_filename, outputs_dir, output_num=str(i))


            job0.addChild(job)


        toil.start(job0)

Running CWL workflows with InplaceUpdateRequirement¶

Some CWL workflows use the InplaceUpdateRequirement feature, which requires that operations on files have visible side effects that Toil's file store cannot support. If you need to run a workflow like this, you can make sure that all of your worker nodes have a shared filesystem, and use the --bypass-file-store option to toil-cwl-runner. This will make it leave all CWL intermediate files on disk and share them between jobs using file paths, instead of storing them in the file store and downloading them when jobs need them.

Toil & CWL Tips¶

See logs for just one job by using the full log file

This requires knowing the job's toil-generated ID, which can be found in the log files.

cat cwltoil.log | grep jobVM1fIs

Grep for full tool commands from toil logs

This gives you a more concise view of the commands being run (note that this information is only available from Toil when running with --logDebug).

pcregrep -M "\[job .*\.cwl.*$\n(.*        .*$\n)*" cwltoil.log
#         ^allows for multiline matching

Find Bams that have been generated for specific step while pipeline is running:

find . | grep -P '^./out_tmpdir.*_MD\.bam$'

See what jobs have been run

cat log/cwltoil.log | grep -oP "\[job .*.cwl\]" | sort | uniq

or:

cat log/cwltoil.log | grep -i "issued job"

Get status of a workflow

$ toil status /home/johnsoni/TEST_RUNS_3/TEST_run/tmp/jobstore-09ae0acc-c800-11e8-9d09-70106fb1697e
<hostname> 2018-10-04 15:01:44,184 MainThread INFO toil.lib.bioio: Root logger is at level 'INFO', 'toil' logger at level 'INFO'.
<hostname> 2018-10-04 15:01:44,185 MainThread INFO toil.utils.toilStatus: Parsed arguments
<hostname> 2018-10-04 15:01:47,081 MainThread INFO toil.utils.toilStatus: Traversing the job graph gathering jobs. This may take a couple of minutes.
Of the 286 jobs considered, there are 179 jobs with children, 107 jobs ready to run, 0 zombie jobs, 0 jobs with services, 0 services, and 0 jobs with log files currently in file:/home/user/jobstore-09ae0acc-c800-11e8-9d09-70106fb1697e.

Toil Stats

You can get run statistics broken down by CWL file. This only works once the workflow is finished:

$ toil stats /path/to/jobstore

This will report resource usage information for all the CWL jobs executed by the workflow.

See Stats Command for an explanation of what the different fields mean.

Understanding toil log files

There is a worker_log.txt file for each Toil job. This file is written to while the job is running, and uploaded at the end if the job finishes or if running at debug log level. If uploaded, the contents are printed to the main log file and transferred to a log file in the --logDir folder.

The new log file will be named something like:

CWLJob_<name of the CWL job>_<attempt number>.log

Standard output/error files will be named like:

<name of the CWL job>.stdout_<attempt number>.log

If you have a workflow revsort.cwl which has a step rev which calls the tool revtool.cwl, the CWL job name ends up being all those parts strung together with .: revsort.cwl.rev.revtool.cwl.

Toil WDL Runner Options¶

--jobStore: Specifies where to keep the Toil state information while running the workflow. Must be accessible from all machines.

-o or --outputDirectory: Specifies the output folder or URI prefix to save workflow output files in. Defaults to a new directory in the current directory.

-m or --outputFile: Specifies a JSON file name or URI to save workflow output values at. Defaults to standard output.

-i or --input: Alternative to the positional argument for the input JSON file, for compatibility with other WDL runners.

--outputDialect: Specifies an output format dialect. Can be cromwell to just return the workflow's output values as JSON or miniwdl to nest that under an outputs key and includes a dir key.

--container: Specifies the container engine to use to run tasks. By default this is auto, which tries Singularity if it is installed and Docker if it isn't. Can also be set to docker or singularity explicitly.

Any number of other Toil options may also be specified. For defined Toil options, see Commandline Options.

Managing Workflow Logs¶

At the default settings, if a WDL task succeeds, the standard output and standard error will be printed in the toil-wdl-runner output, unless they are captured by the workflow (with the stdout() and stderr() WDL built-in functions). If a WDL task fails, they will be printed whether they were meant to be captured or not. Complete logs from Toil for failed jobs will also be printed.

If you would like to save the logs organized by WDL task, you can use the --writeLogs or --writeLogsGzip options to specify a directory where the log files should be saved. Log files will be named after the same dotted, hierarchical workflow and task names used to set values from the input JSON, except that scatters will add an additional numerical component. In addition to the logs for WDL tasks, Toil job logs for failed jobs will also appear here when running at the default log level.

For example, if you run:

toil-wdl-runner --writeLogs logs https://raw.githubusercontent.com/DataBiosphere/toil/36b54c45e8554ded5093bcdd03edb2f6b0d93887/src/toil/test/wdl/miniwdl_self_test/self_test.wdl https://raw.githubusercontent.com/DataBiosphere/toil/36b54c45e8554ded5093bcdd03edb2f6b0d93887/src/toil/test/wdl/miniwdl_self_test/inputs.json

You will end up with a logs/ directory containing:

hello_caller.0.hello.stderr_000.log
hello_caller.1.hello.stderr_000.log
hello_caller.2.hello.stderr_000.log

The final number is a sequential counter: if a step has to be retried, or if you run the workflow multiple times without clearing out the logs directory, it will increment.

Using the UCSC Genomics Institute Tutorial¶

The UCSC Genomics Institute (home of the Toil project) has a tutorial on writing WDL workflows with Toil. You can follow this tutorial to be walked through writing your own WDL workflow with Toil. They also have tips on debugging WDL workflows with Toil.

These tutorials and tips are aimed at users looking to run WDL workflows with Toil in a Slurm environment, but they can also apply in other situations.

Using the Official WDL tutorials¶

You can also learn to write WDL workflows for Toil by following the official WDL tutorials.

When you reach the point of executing your workflow, instead of running with Cromwell:

java -jar Cromwell.jar run myWorkflow.wdl --inputs myWorkflow_inputs.json

you can instead run with toil-wdl-runner:

toil-wdl-runner myWorkflow.wdl --inputs myWorkflow_inputs.json

Using the Learn WDL Video Tutorials¶

For people who prefer video tutorials, Lynn Langit has a Learn WDL video course that will teach you how to write and run WDL workflows. The course is taught using Cromwell, but Toil should also be compatible with the course's workflows.

WDL Specifications¶

WDL language specifications can be found here: https://github.com/openwdl/wdl/blob/main/versions/1.1/SPEC.md

Toil is not yet fully conformant with the WDL specification, but it inherits most of the functionality of MiniWDL.

Job Store¶

The job store is a storage abstraction which contains all of the information used in a Toil run. This centralizes all of the files used by jobs in the workflow and also the details of the progress of the run. If a workflow crashes or fails, the job store contains all of the information necessary to resume with minimal repetition of work.

Several different job stores are supported, including the file job store and cloud job stores. For information on developing job stores, see Job Store API.

File Job Store¶

The file job store is for use locally, and keeps the workflow information in a directory on the machine where the workflow is launched. This is the simplest and most convenient job store for testing or for small runs.

For an example that uses the file job store, see Running a basic CWL workflow.

Cloud Job Stores¶

Toil currently supports the following cloud storage systems as job stores:

These use cloud buckets to house all of the files. This is useful if there are several different worker machines all running jobs that need to access the job store.

Batch System¶

A Toil batch system is either a local single-machine (one computer) or a currently supported cluster of computers (lsf, mesos, slurm, torque, htcondor, or grid_engine) These environments manage individual worker nodes under a leader node to process the work required in a workflow. The leader and its workers all coordinate their tasks and files through a centralized job store location.

See Batch System API for a more detailed description of different batch systems, or information on developing batch systems.

Provisioner¶

The Toil provisioner provides a tool set for running a Toil workflow on a particular cloud platform.

The Toil Cluster Utilities are command line tools used to provision nodes in your desired cloud platform. They allows you to launch nodes, ssh to the leader, and rsync files back and forth.

For detailed instructions for using the provisioner see Running in AWS or Running in Google Compute Engine (GCE).

The Config File¶

Instead of changing the arguments on the command line, Toil offers support for using a configuration file.

Options will be applied with priority:

You can manually generate an example configuration file to a path you select. To generate a configuration file, run:

$ toil config [filename].yaml

Then uncomment options as necessary and change/provide new values.

After editing the config file, you can run Toil with its settings by passing it on the command line:

$ python3 example.py --config=[filename].yaml

Alternatively, you can edit the default config file, which is located at $HOME/.toil/default.yaml

If CLI options are used in addition to the configuration file, the CLI options will overwrite the configuration file options. For example:

$ python3 example.py --config=[filename].yaml --defaultMemory 80Gi

This will result in a default memory per job of 80GiB no matter what is in the configuration file provided.

The Job Store¶

Running Toil workflows requires a file path or URL to a central location for all of the intermediate files for the workflow: the job store. For toil-cwl-runner and toil-wdl-runner a job store can often be selected automatically or can be specified with the --jobStore option; Toil Python workflows generally require the job store as a positional command line argument. To use the Python quickstart example, if you're on a node that has a large /scratch volume, you can specify that the jobstore be created there by executing: python3 HelloWorld.py /scratch/my-job-store, or more explicitly, python3 HelloWorld.py file:/scratch/my-job-store.

Syntax for specifying different job stores:

Different types of job store options can be found below.

Commandline Options¶

Core Toil Options Options to specify the location of the Toil workflow and turn on stats collation about the performance of jobs.

Logging Options Toil hides stdout and stderr by default except in case of job failure. Log levels in toil are based on priority from the logging module:

Batch System Options

Data Storage Options Allows configuring Toil's data storage.

Autoscaling Options Allows the specification of the minimum and maximum number of nodes in an autoscaled cluster, as well as parameters to control the level of provisioning.

Service Options Allows the specification of the maximum number of service jobs in a cluster. By keeping this limited we can avoid nodes occupied with services causing deadlocks. (Not for CWL).

Resource Options The options to specify default cores/memory requirements (if not specified by the jobs themselves), and to limit the total amount of memory/cores requested from the batch system.

Options for rescuing/killing/restarting jobs. The options for jobs that either run too long/fail or get lost (some batch systems have issues!).

Log Management Options

Miscellaneous Options

Debug Options Debug options for finding problems or helping with testing.

Restart Option¶

In the event of failure, Toil can resume the pipeline by adding the argument --restart and rerunning the workflow. Toil Python workflows (but not CWL or WDL workflows) can even be edited and resumed, which is useful for development or troubleshooting.

Running Workflows with Services¶

Toil supports jobs, or clusters of jobs, that run as services to other accessor jobs. Example services include server databases or Apache Spark Clusters. As service jobs exist to provide services to accessor jobs their runtime is dependent on the concurrent running of their accessor jobs. The dependencies between services and their accessor jobs can create potential deadlock scenarios, where the running of the workflow hangs because only service jobs are being run and their accessor jobs can not be scheduled because of too limited resources to run both simultaneously. To cope with this situation Toil attempts to schedule services and accessors intelligently, however to avoid a deadlock with workflows running service jobs it is advisable to use the following parameters:

Specifying these parameters so that at a maximum cluster size there will be sufficient resources to run accessors in addition to services will ensure that such a deadlock can not occur.

If too low a limit is specified then a deadlock can occur in which toil can not schedule sufficient service jobs concurrently to complete the workflow. Toil will detect this situation if it occurs and throw a toil.DeadlockException exception. Increasing the cluster size and these limits will resolve the issue.

Setting Options directly in a Python Workflow¶

It's good to remember that commandline options can be overridden in the code of a Python workflow. For example, toil.job.Job.Runner.getDefaultOptions() can be used to get the default Toil options, ignoring what was passed on the command line. In this example, this is used to ignore command-line options and always run with the "./toilWorkflow" directory as the jobstore:

options = Job.Runner.getDefaultOptions("./toilWorkflow") # Get the options object
with Toil(options) as toil:


    toil.start(Job())  # Run the root job

However, each option can be explicitly set within the workflow by modifying the options object. In this example, we are setting logLevel = "DEBUG" (all log statements are shown) and clean="ALWAYS" (always delete the jobstore) like so:

options = Job.Runner.getDefaultOptions("./toilWorkflow") # Get the options object
options.logLevel = "DEBUG" # Set the log level to the debug level.
options.clean = "ALWAYS" # Always delete the jobStore after a run
with Toil(options) as toil:


    toil.start(Job())  # Run the root job

However, the usual incantation is to accept commandline args from the user with the following:

parser = Job.Runner.getDefaultArgumentParser() # Get the parser
options = parser.parse_args() # Parse user args to create the options object
with Toil(options) as toil:


    toil.start(Job())  # Run the root job

We can also have code in the workflow to overwrite user supplied arguments:

parser = Job.Runner.getDefaultArgumentParser() # Get the parser
options = parser.parse_args() # Parse user args to create the options object
options.logLevel = "DEBUG" # Set the log level to the debug level.
options.clean = "ALWAYS" # Always delete the jobStore after a run
with Toil(options) as toil:


    toil.start(Job())  # Run the root job

Stats Command¶

To use the stats command, a workflow must first be run using the --stats option. Using this command makes certain that toil does not delete the job store, no matter what other options are specified (i.e. normally the option --clean=always would delete the job store, but --stats will override this).

Running an Example¶

We can run an example workflow and record stats:

python3 discoverfiles.py file:my-jobstore --stats

Where discoverfiles.py is the following:

import os
from toil.common import Toil
from toil.job import Job
import math
import time
from multiprocessing import Process
def think(seconds):


    start = time.time()


    while time.time() - start < seconds:


        # Use CPU


        math.sqrt(123456)
class TimeWaster(Job):


    def __init__(self, time_to_think, time_to_waste, space_to_waste, *args, **kwargs):


        self.time_to_think = time_to_think


        self.time_to_waste = time_to_waste


        self.space_to_waste = space_to_waste


        super().__init__(*args, **kwargs)


    def run(self, fileStore):


        # Waste some space


        file_path = fileStore.getLocalTempFile()


        with open(file_path, 'w') as stream:


            for i in range(self.space_to_waste):


                stream.write("X")


        


        # Do some "useful" compute


        processes = []


        for core_number in range(max(1, self.cores)):


            # Use all the assigned cores to think


            p = Process(target=think, args=(self.time_to_think,))


            p.start()


            processes.append(p)


        for p in processes:


            p.join()


        # Also waste some time


        time.sleep(self.time_to_waste)
def main():


    options = Job.Runner.getDefaultArgumentParser().parse_args()


    job1 = TimeWaster(0, 0, 0, displayName='doNothing')


    job2 = TimeWaster(10, 0, 4096, displayName='efficientJob')


    job3 = TimeWaster(10, 0, 1024, cores=4, displayName='multithreadedJob')


    job4 = TimeWaster(1, 9, 65536, displayName='inefficientJob')


    


    job1.addChild(job2)


    job1.addChild(job3)


    job3.addChild(job4)


    with Toil(options) as toil:


        if not toil.options.restart:


            toil.start(job1)


        else:


            toil.restart()
if __name__ == '__main__':


    main()

Notice the displayName key, which can rename a job, giving it an alias when it is finally displayed in stats.

Displaying Stats¶

To see the runtime and resources used for each job when it was run, type

toil stats file:my-jobstore

This should output something like the following:

Batch System: single_machine
Default Cores: 1  Default Memory: 2097152KiB
Max Cores: unlimited
Local CPU Time: 55.54 core·s  Overall Runtime: 26.23 s
Worker


    Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


        3 |     0.34   10.83   10.80   21.23   32.40 |     0.33   10.43   17.94   43.07   53.83 |     0.01    0.40   14.08   41.85   42.25 |  177168Ki 179312Ki 178730Ki 179712Ki 536192Ki |      0Ki     4Ki    22Ki    64Ki    68Ki
Job


 Worker Jobs  |     min    med    ave    max


              |       1      1 1.3333      2


    Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


        4 |     0.33   10.83    8.10   10.85   32.38 |     0.33   10.43   13.46   41.70   53.82 |     0.01    1.68    2.78    9.02   11.10 |  177168Ki 179488Ki 178916Ki 179696Ki 715664Ki |      0Ki     4Ki    18Ki    64Ki    72Ki


 multithreadedJob


    Total Cores: 4.0


    Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


        1 |    10.85   10.85   10.85   10.85   10.85 |    41.70   41.70   41.70   41.70   41.70 |     1.68    1.68    1.68    1.68    1.68 |  179488Ki 179488Ki 179488Ki 179488Ki 179488Ki |      4Ki     4Ki     4Ki     4Ki     4Ki


 efficientJob


    Total Cores: 1.0


    Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


        1 |    10.83   10.83   10.83   10.83   10.83 |    10.43   10.43   10.43   10.43   10.43 |     0.40    0.40    0.40    0.40    0.40 |  179312Ki 179312Ki 179312Ki 179312Ki 179312Ki |      4Ki     4Ki     4Ki     4Ki     4Ki


 inefficientJob


    Total Cores: 1.0


    Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


        1 |    10.38   10.38   10.38   10.38   10.38 |     1.36    1.36    1.36    1.36    1.36 |     9.02    9.02    9.02    9.02    9.02 |  179696Ki 179696Ki 179696Ki 179696Ki 179696Ki |     64Ki    64Ki    64Ki    64Ki    64Ki


 doNothing


    Total Cores: 1.0


    Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


        1 |     0.33    0.33    0.33    0.33    0.33 |     0.33    0.33    0.33    0.33    0.33 |     0.01    0.01    0.01    0.01    0.01 |  177168Ki 177168Ki 177168Ki 177168Ki 177168Ki |      0Ki     0Ki     0Ki     0Ki     0Ki

This report gives information on the resources used by your workflow. Note that right now it does NOT track CPU and memory used inside Docker containers, only Singularity containers.

There are three parts to this report.

Overall Summary¶

At the top is a section with overall summary statistics for the run:

Batch System: single_machine
Default Cores: 1  Default Memory: 2097152KiB
Max Cores: unlimited
Local CPU Time: 55.54 core·s  Overall Runtime: 26.23 s

This lists some important the settings for the Toil batch system that actually executed jobs. It also lists:

These latter two numbers don't count some startup/shutdown time spent loading and saving files, so you still may want to use the time shell built-in to time your Toil runs overall.

Worker Summary¶

After the overall summary, there is a section with statistics about the Toil worker processes, which Toil used to execute your workflow's jobs:

Worker


    Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


        3 |     0.34   10.83   10.80   21.23   32.40 |     0.33   10.43   17.94   43.07   53.83 |     0.01    0.40   14.08   41.85   42.25 |  177168Ki 179312Ki 178730Ki 179712Ki 536192Ki |      0Ki     4Ki    22Ki    64Ki    68Ki

Job Breakdown¶

Finally, there is the breakdown of resource usage by jobs. This starts with a table summarizing the counts of jobs that ran on each worker:

Job


 Worker Jobs  |     min    med    ave    max


              |       1      1 1.3333      2

In this example, most of the workers ran one job each, but one worker managed to run two jobs, via chaining. (Jobs will chain when a job has only one dependant job, which in turn depends on only that first job, and the second job needs no more resources than the first job did.)

Next, we have statistics for resource usage over all jobs together:

Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


    n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


    4 |     0.33   10.83    8.10   10.85   32.38 |     0.33   10.43   13.46   41.70   53.82 |     0.01    1.68    2.78    9.02   11.10 |  177168Ki 179488Ki 178916Ki 179696Ki 715664Ki |      0Ki     4Ki    18Ki    64Ki    72Ki

And finally, for each kind of job (as determined by the job's displayName), we have statistics summarizing the resources used by the instances of that kind of job:

multithreadedJob


   Total Cores: 4.0


   Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


       n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


       1 |    10.85   10.85   10.85   10.85   10.85 |    41.70   41.70   41.70   41.70   41.70 |     1.68    1.68    1.68    1.68    1.68 |  179488Ki 179488Ki 179488Ki 179488Ki 179488Ki |      4Ki     4Ki     4Ki     4Ki     4Ki
efficientJob


   Total Cores: 1.0


   Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


       n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


       1 |    10.83   10.83   10.83   10.83   10.83 |    10.43   10.43   10.43   10.43   10.43 |     0.40    0.40    0.40    0.40    0.40 |  179312Ki 179312Ki 179312Ki 179312Ki 179312Ki |      4Ki     4Ki     4Ki     4Ki     4Ki
inefficientJob


   Total Cores: 1.0


   Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


       n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


       1 |    10.38   10.38   10.38   10.38   10.38 |     1.36    1.36    1.36    1.36    1.36 |     9.02    9.02    9.02    9.02    9.02 |  179696Ki 179696Ki 179696Ki 179696Ki 179696Ki |     64Ki    64Ki    64Ki    64Ki    64Ki
doNothing


   Total Cores: 1.0


   Count |                           Real Time (s)* |                        CPU Time (core·s) |                        CPU Wait (core·s) |                                    Memory (B) |                                 Disk (B)


       n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |       min      med      ave      max    total |      min     med     ave     max   total


       1 |     0.33    0.33    0.33    0.33    0.33 |     0.33    0.33    0.33    0.33    0.33 |     0.01    0.01    0.01    0.01    0.01 |  177168Ki 177168Ki 177168Ki 177168Ki 177168Ki |      0Ki     0Ki     0Ki     0Ki     0Ki

For each job, we first list its name, and then the total cores that it asked for, summed across all instances of it. Then we show a table of statistics.

Here the * marker in the table headers becomes relevant; it shows that jobs are being sorted by the median of the real time used. You can control this with the --sortCategory option.

The columns meanings are the same as for the workers:

Example Cleanup¶

Once we're done looking at the stats, we can clean up the job store by running:

toil clean file:my-jobstore

Status Command¶

Continuing the example from the stats section above, if we ran our workflow with the command

python3 discoverfiles.py file:my-jobstore --stats

We could interrogate our jobstore with the status command, for example:

toil status file:my-jobstore

If the run was successful, this would not return much valuable information, something like

2018-01-11 19:31:29,739 - toil.lib.bioio - INFO - Root logger is at level 'INFO', 'toil' logger at level 'INFO'.
2018-01-11 19:31:29,740 - toil.utils.toilStatus - INFO - Parsed arguments
2018-01-11 19:31:29,740 - toil.utils.toilStatus - INFO - Checking if we have files for Toil
The root job of the job store is absent, the workflow completed successfully.

Otherwise, the status command should return the following:

Clean Command¶

If a Toil pipeline didn't finish successfully, or was run using --clean=always or --stats, the job store will exist until it is deleted. toil clean <jobStore> ensures that all artifacts associated with a job store are removed. This is particularly useful for deleting AWS job stores, which reserves an SDB domain as well as an S3 bucket.

The deletion of the job store can be modified by the --clean argument, and may be set to always, onError, never, or onSuccess (default).

Temporary directories where jobs are running can also be saved from deletion using the --cleanWorkDir, which has the same options as --clean. This option should only be run when debugging, as intermediate jobs will fill up disk space.

Kill Command¶

To kill all currently running jobs for a given jobstore, use the command

toil kill file:my-jobstore

Introspecting the Jobstore¶

Note: Currently these features are only implemented for use locally (single machine) with the fileJobStore.

To view what files currently reside in the jobstore, run the following command:

$ toil debug-file file:path-to-jobstore-directory \


      --listFilesInJobStore

When run from the commandline, this should generate a file containing the contents of the job store (in addition to displaying a series of log messages to the terminal). This file is named "jobstore_files.txt" by default and will be generated in the current working directory.

If one wishes to copy any of these files to a local directory, one can run for example:

$ toil debug-file file:path-to-jobstore \


      --fetch overview.txt *.bam *.fastq \


      --localFilePath=/home/user/localpath

To fetch overview.txt, and all .bam and .fastq files. This can be used to recover previously used input and output files for debugging or reuse in other workflows, or use in general debugging to ensure that certain outputs were imported into the jobStore.

Stats and Status¶

See Status Command for more about gathering statistics about job success, runtime, and resource usage from workflows.

Using a Python debugger¶

If you execute a workflow using the --debugWorker flag, Toil will not fork in order to run jobs, which means you can either use pdb, or an IDE that supports debugging Python as you would normally. Note that the --debugWorker flag will only work with the single_machine batch system (the default), and not any of the custom job schedulers.

Managing a Cluster of Virtual Machines (Provisioning)¶

Toil can launch and manage a cluster of virtual machines to run using the provisioner to run a workflow distributed over several nodes. The provisioner also has the ability to automatically scale up or down the size of the cluster to handle dynamic changes in computational demand (autoscaling). Currently we have working provisioners with AWS and GCE (Azure support has been deprecated).

Toil uses Kubernetes as the Batch System.

See here for instructions for Running in AWS.

See here for instructions for Running in Google Compute Engine (GCE).

Toil offers a suite of commands for using the provisioners to manage clusters.

Toil Cluster Utilities¶

In addition to the generic Toil Utilities, there are several utilities used for starting and managing a Toil cluster using the AWS or GCE provisioners. They are installed via the [aws] or [google] extra. For installation details see Toil Provisioner.

The toil cluster subcommands are:

For information on a specific utility, run it with the --help option:

toil launch-cluster --help

The cluster utilities can be used for Running in Google Compute Engine (GCE) and Running in AWS.

TIP:

NOTE:

Launch-Cluster Command¶

Running toil launch-cluster starts up a leader for a cluster. Workers can be added to the initial cluster by specifying the -w option. An example would be

$ toil launch-cluster my-cluster \


      --leaderNodeType t2.small -z us-west-2a \


      --keyPairName your-AWS-key-pair-name \


      --nodeTypes m3.large,t2.micro -w 1,4

Options are listed below. These can also be displayed by running

$ toil launch-cluster --help

launch-cluster's main positional argument is the clusterName. This is simply the name of your cluster. If it does not exist yet, Toil will create it for you.

Launch-Cluster Options

Logging Options

Ssh-Cluster Command¶

Toil provides the ability to ssh into the leader of the cluster. This can be done as follows:

$ toil ssh-cluster CLUSTER-NAME-HERE

This will open a shell on the Toil leader and is used to start an Running a Workflow with Autoscaling run. Issues with docker prevent using screen and tmux when sshing the cluster (The shell doesn't know that it is a TTY which prevents it from allocating a new screen session). This can be worked around via

$ script
$ screen

Simply running screen within script will get things working properly again.

Finally, you can execute remote commands with the following syntax:

$ toil ssh-cluster CLUSTER-NAME-HERE remoteCommand

It is not advised that you run your Toil workflow using remote execution like this unless a tool like nohup is used to ensure the process does not die if the SSH connection is interrupted.

For an example usage, see Running a Workflow with Autoscaling.

Rsync-Cluster Command¶

The most frequent use case for the rsync-cluster utility is deploying your workflow code to the Toil leader. Note that the syntax is the same as traditional rsync with the exception of the hostname before the colon. This is not needed in toil rsync-cluster since the hostname is automatically determined by Toil.

Here is an example of its usage:

$ toil rsync-cluster CLUSTER-NAME-HERE \


   ~/localFile :/remoteDestination

Destroy-Cluster Command¶

The destroy-cluster command is the advised way to get rid of any Toil cluster launched using the Launch-Cluster Command command. It ensures that all attached nodes, volumes, security groups, etc. are deleted. If a node or cluster is shut down using Amazon's online portal residual resources may still be in use in the background. To delete a cluster run

$ toil destroy-cluster CLUSTER-NAME-HERE

Storage (Toil jobStore)¶

Toil can make use of cloud storage such as AWS or Google buckets to take care of storage needs.

This is useful when running Toil in single machine mode on any cloud platform since it allows you to make use of their integrated storage systems.

For an overview of the job store see Job Store.

For instructions configuring a particular job store see:

Running on Kubernetes¶

Kubernetes is a very popular container orchestration tool that has become a de facto cross-cloud-provider API for accessing cloud resources. Major cloud providers like Amazon, Microsoft, Kubernetes owner Google, and DigitalOcean have invested heavily in making Kubernetes work well on their platforms, by writing their own deployment documentation and developing provider-managed Kubernetes-based products. Using minikube, Kubernetes can even be run on a single machine.

Toil supports running Toil workflows against a Kubernetes cluster, either in the cloud or deployed on user-owned hardware.

Preparing your Kubernetes environment¶

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:


  namespace: YOUR_NAMESPACE_HERE


  name: toil-user
rules:
- apiGroups: ["*"]


  resources: ["*"]


  verbs: ["explain", "get", "watch", "list", "describe", "logs", "attach", "exec", "port-forward", "proxy", "cp", "auth"]
- apiGroups: ["batch"]


  resources: ["*"]


  verbs: ["get", "watch", "list", "create", "run", "set", "delete"]
- apiGroups: [""]


  resources: ["secrets", "pods", "pods/attach", "podtemplates", "configmaps", "events", "services"]


  verbs: ["patch", "get", "update", "watch", "list", "create", "run", "set", "delete", "exec"]
- apiGroups: [""]


  resources: ["pods", "pods/log"]


  verbs: ["get", "list"]
- apiGroups: [""]


  resources: ["pods/exec"]


  verbs: ["create"]

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:


  name: node-reader
rules:
- apiGroups: [""]


  resources: ["nodes"]


  verbs: ["get", "list", "describe"]
- apiGroups: [""]


  resources: ["namespaces"]


  verbs: ["get", "list", "describe"]
- apiGroups: ["metrics.k8s.io"]


  resources: ["*"]


  verbs: ["*"]

apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:


  name: toil-developer-member


  namespace: toil
subjects:
- kind: User


  name: YOUR_KUBERNETES_USERNAME_HERE


  apiGroup: rbac.authorization.k8s.io
- kind: ServiceAccount


  name: YOUR_SERVICE_ACCOUNT_NAME_HERE


  namespace: YOUR_NAMESPACE_HERE
roleRef:


  kind: Role


  name: toil-user


  apiGroup: rbac.authorization.k8s.io

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:


  name: read-nodes
subjects:
- kind: User


  name: YOUR_KUBERNETES_USERNAME_HERE


  apiGroup: rbac.authorization.k8s.io
- kind: ServiceAccount


  name: YOUR_SERVICE_ACCOUNT_NAME_HERE


  namespace: YOUR_NAMESPACE_HERE
roleRef:


  kind: ClusterRole


  name: node-reader


  apiGroup: rbac.authorization.k8s.io

AWS Job Store for Kubernetes¶

Currently, the only job store, which is what Toil uses to exchange data between jobs, that works with jobs running on Kubernetes is the AWS Job Store. This requires that the Toil leader and Kubernetes jobs be able to connect to and use Amazon S3 and Amazon SimpleDB. It also requires that you have an Amazon Web Services account.

https://console.aws.amazon.com/iam/home?region=us-west-1#/home

$ aws configure

[default]
aws_access_key_id =  BLAH
aws_secret_access_key =  blahblahblah

$ cd ~/.aws

$ kubectl create secret generic aws-credentials --from-file credentials

Configuring Toil for your Kubernetes environment¶

To configure your workflow to run on Kubernetes, you will have to configure several environment variables, in addition to passing the --batchSystem kubernetes option. Doing the research to figure out what values to give these variables may require talking to your cluster provider.

Note that Docker containers cannot be run inside of unprivileged Kubernetes pods (which are themselves containers). The Docker daemon does not (yet) support this. Other tools, such as Singularity in its user-namespace mode, are able to run containers from within containers. If using Singularity to run containerized tools, and you want downloaded container images to persist between Toil jobs, some setup may be required:

On non-Toil managed clusters: You will also want to set TOIL_KUBERNETES_HOST_PATH, and make sure that Singularity is downloading its containers under the Toil work directory (/var/lib/toil by default) by setting SINGULARITY_CACHEDIR.

On Toil-managed clusters: On clusters created with the launch-cluster command, no setup is required. TOIL_KUBERNETES_HOST_PATH is already set to /var/lib/toil. SINGULARITY_CACHEDIR is set to /var/lib/toil/singularity which is a shared location; however, you may need to implement Singularity locking as shown below or change the Singularity cache location to somewhere else.

If using toil-wdl-runner, all the necessary locking for Singularity is already in place and no work should be necessary. Else, for both Toil managed and non-Toil managed clusters, you will need to make sure that no two jobs try to download the same container at the same time; Singularity has no synchronization or locking around its cache, but the cache is also not safe for simultaneous access by multiple Singularity invocations. Some Toil workflows use their own custom workaround logic for this problem; for example, see this section in toil-wdl-runner.

Running workflows¶

To run the workflow, you will need to run the Toil leader process somewhere. It can either be run inside Kubernetes as a Kubernetes job, or outside Kubernetes as a normal command.

Option 1: Running the Leader Inside Kubernetes¶

Once you have determined a set of environment variable values for your workflow run, write a YAML file that defines a Kubernetes job to run your workflow with that configuration. Some configuration items (such as your username, and the name of your AWS credentials secret) need to be written into the YAML so that they can be used from the leader as well.

Note that the leader pod will need your workflow, its other dependencies, and Toil all installed. An easy way to get Toil installed is to start with the Toil appliance image for the version of Toil you want to use. In this example, we use quay.io/ucsc_cgl/toil:5.5.0.

Here's an example YAML file to run a test workflow:

apiVersion: batch/v1
kind: Job
metadata:


  # It is good practice to include your username in your job name.


  # Also specify it in TOIL_KUBERNETES_OWNER


  name: demo-user-toil-test
# Do not try and rerun the leader job if it fails
spec:


 backoffLimit: 0


 template:


   spec:


     # Do not restart the pod when the job fails, but keep it around so the


     # log can be retrieved


     restartPolicy: Never


     volumes:


     - name: aws-credentials-vol


       secret:


         # Make sure the AWS credentials are available as a volume.


         # This should match TOIL_AWS_SECRET_NAME


         secretName: aws-credentials


     # You may need to replace this with a different service account name as


     # appropriate for your cluster.


     serviceAccountName: default


     containers:


     - name: main


       image: quay.io/ucsc_cgl/toil:5.5.0


       env:


       # Specify your username for inclusion in job names


       - name: TOIL_KUBERNETES_OWNER


         value: demo-user


       # Specify where to find the AWS credentials to access the job store with


       - name: TOIL_AWS_SECRET_NAME


         value: aws-credentials


       # Specify where per-host caches should be stored, on the Kubernetes hosts.


       # Needs to be set for Toil's caching to be efficient.


       - name: TOIL_KUBERNETES_HOST_PATH


         value: /data/scratch


       volumeMounts:


       # Mount the AWS credentials volume


       - mountPath: /root/.aws


         name: aws-credentials-vol


       resources:


         # Make sure to set these resource limits to values large enough


         # to accommodate the work your workflow does in the leader


         # process, but small enough to fit on your cluster.


         #


         # Since no request values are specified, the limits are also used


         # for the requests.


         limits:


           cpu: 2


           memory: "4Gi"


           ephemeral-storage: "10Gi"


       command:


       - /bin/bash


       - -c


       - |


         # This Bash script will set up Toil and the workflow to run, and run them.


         set -e


         # We make sure to create a work directory; Toil can't hot-deploy a


         # Python file from the root of the filesystem, which is where we start.


         mkdir /tmp/work


         cd /tmp/work


         # We make a virtual environment to allow workflow dependencies to be


         # hot-deployed.


         #


         # We don't really make use of it in this example, but for workflows


         # that depend on PyPI packages we will need this.


         #


         # We use --system-site-packages so that the Toil installed in the


         # appliance image is still available.


         virtualenv --python python3 --system-site-packages venv


         . venv/bin/activate


         # Now we install the workflow. Here we're using a demo workflow


         # from Toil itself.


         wget https://raw.githubusercontent.com/DataBiosphere/toil/releases/4.1.0/src/toil/test/docs/scripts/tutorial_helloworld.py


         # Now we run the workflow. We make sure to use the Kubernetes batch


         # system and an AWS job store, and we set some generally useful


         # logging options. We also make sure to enable caching.


         python3 tutorial_helloworld.py \


             aws:us-west-2:demouser-toil-test-jobstore \


             --batchSystem kubernetes \


             --realTimeLogging \


             --logInfo

You can save this YAML as leader.yaml, and then run it on your Kubernetes installation with:

$ kubectl apply -f leader.yaml

To monitor the progress of the leader job, you will want to read its logs. If you are using a Kubernetes dashboard such as k9s, you can simply find the pod created for the job in the dashboard, and view its logs there. If not, you will need to locate the pod by hand.

Monitoring and Debugging Kubernetes Jobs and Pods¶

The following techniques are most useful for looking at the pod which holds the Toil leader, but they can also be applied to individual Toil jobs on Kubernetes, even when the leader is outside the cluster.

Kubernetes names pods for jobs by appending a short random string to the name of the job. You can find the name of the pod for your job by doing:

$ kubectl get pods | grep demo-user-toil-test
demo-user-toil-test-g5496                                         1/1     Running     0          2m

Assuming you have set TOIL_KUBERNETES_OWNER correctly, you should be able to find all of your workflow's pods by searching for your username:

$ kubectl get pods | grep demo-user

If the status of a pod is anything other than Pending, you will be able to view its logs with:

$ kubectl logs demo-user-toil-test-g5496

This will dump the pod's logs from the beginning to now and terminate. To follow along with the logs from a running pod, add the -f option:

$ kubectl logs -f demo-user-toil-test-g5496

A status of ImagePullBackoff suggests that you have requested to use an image that is not available. Check the image section of your YAML if you are looking at a leader, or the value of TOIL_APPLIANCE_SELF if you are delaying with a worker job. You also might want to check your Kubernetes node's Internet connectivity and DNS function; in Kubernetes, DNS depends on system-level pods which can be terminated or evicted in cases of resource oversubscription, just like user workloads.

If your pod seems to be stuck Pending, ContainerCreating, you can get information on what is wrong with it by using kubectl describe pod:

$ kubectl describe pod demo-user-toil-test-g5496

Pay particular attention to the Events: section at the end of the output. An indication that a job is too big for the available nodes on your cluster, or that your cluster is too busy for your jobs, is FailedScheduling events:

Type     Reason            Age                  From               Message
----     ------            ----                 ----               -------
Warning  FailedScheduling  13s (x79 over 100m)  default-scheduler  0/4 nodes are available: 1 Insufficient cpu, 1 Insufficient ephemeral-storage, 4 Insufficient memory.

If a pod is running but seems to be behaving erratically, or seems stuck, you can shell into it and look around:

$ kubectl exec -ti demo-user-toil-test-g5496 /bin/bash

One common cause of stuck pods is attempting to use more memory than allowed by Kubernetes (or by the Toil job's memory resource requirement), but in a way that does not trigger the Linux OOM killer to terminate the pod's processes. In these cases, the pod can remain stuck at nearly 100% memory usage more or less indefinitely, and attempting to shell into the pod (which needs to start a process within the pod, using some of its memory) will fail. In these cases, the recommended solution is to kill the offending pod and increase its (or its Toil job's) memory requirement, or reduce its memory needs by adapting user code.

When Things Go Wrong¶

The Toil Kubernetes batch system includes cleanup code to terminate worker jobs when the leader shuts down. However, if the leader pod is removed by Kubernetes, is forcibly killed or otherwise suffers a sudden existence failure, it can go away while its worker jobs live on. It is not recommended to restart a workflow in this state, as jobs from the previous invocation will remain running and will be trying to modify the job store concurrently with jobs from the new invocation.

To clean up dangling jobs, you can use the following snippet:

$ kubectl get jobs | grep demo-user | cut -f1 -d' ' | xargs -n10 kubectl delete job

This will delete all jobs with demo-user's username in their names, in batches of 10. You can also use the UUID that Toil assigns to a particular workflow invocation in the filter, to clean up only the jobs pertaining to that workflow invocation.

Option 2: Running the Leader Outside Kubernetes¶

If you don't want to run your Toil leader inside Kubernetes, you can run it locally instead. This can be useful when developing a workflow; files can be hot-deployed from your local machine directly to Kubernetes. However, your local machine will have to have (ideally role-assumption- and MFA-free) access to AWS, and access to Kubernetes. Real time logging will not work unless your local machine is able to listen for incoming UDP packets on arbitrary ports on the address it uses to contact the IPv4 Internet; Toil does no NAT traversal or detection.

Note that if you set TOIL_WORKDIR when running your workflow like this, it will need to be a directory that exists both on the host and in the Toil appliance.

Here is an example of running our test workflow leader locally, outside of Kubernetes:

$ export TOIL_KUBERNETES_OWNER=demo-user  # This defaults to your local username if not set
$ export TOIL_AWS_SECRET_NAME=aws-credentials
$ export TOIL_KUBERNETES_HOST_PATH=/data/scratch
$ virtualenv --python python3 --system-site-packages venv
$ . venv/bin/activate
$ wget https://raw.githubusercontent.com/DataBiosphere/toil/releases/4.1.0/src/toil/test/docs/scripts/tutorial_helloworld.py
$ python3 tutorial_helloworld.py \


      aws:us-west-2:demouser-toil-test-jobstore \


      --batchSystem kubernetes \


      --realTimeLogging \


      --logInfo

Running CWL Workflows¶

Running CWL workflows on Kubernetes can be challenging, because executing CWL can require toil-cwl-runner to orchestrate containers of its own, within a Kubernetes job running in the Toil appliance container.

Normally, running a CWL workflow should Just Work, as long as the workflow's Docker containers are able to be executed with Singularity, your Kubernetes cluster does not impose extra capability-based confinement (i.e. SELinux, AppArmor) that interferes with Singularity's use of user-mode namespaces, and you make sure to configure Toil so that its workers know where to store their data within the Kubernetes pods (which would be done for you if using a Toil-managed cluster). For example, you should be able to run a CWL workflow like this:

$ export TOIL_KUBERNETES_OWNER=demo-user  # This defaults to your local username if not set
$ export TOIL_AWS_SECRET_NAME=aws-credentials
$ export TOIL_KUBERNETES_HOST_PATH=/data/scratch
$ virtualenv --python python3 --system-site-packages venv
$ . venv/bin/activate
$ pip install toil[kubernetes,cwl]==5.8.0
$ toil-cwl-runner  \


     --jobStore  aws:us-west-2:demouser-toil-test-jobstore \


     --batchSystem kubernetes \


     --realTimeLogging \


     --logInfo \


     --disableCaching \


     path/to/cwl/workflow \


     path/to/cwl/input/object

Additional cwltool options that your workflow might require, such as --no-match-user, can be passed to toil-cwl-runner, which inherits most cwltool options.

AppArmor and Singularity¶

Kubernetes clusters based on Ubuntu hosts often will have AppArmor enabled on the host. AppArmor is a capability-based security enhancement system that integrates with the Linux kernel to enforce lists of things which programs may or may not do, called profiles. For example, an AppArmor profile could be applied to a web server process to stop it from using the mount() system call to manipulate the filesystem, because it has no business doing that under normal circumstances but might attempt to do it if compromised by hackers.

Kubernetes clusters also often use Docker as the backing container runtime, to run pod containers. When AppArmor is enabled, Docker will load an AppArmor profile and apply it to all of its containers by default, with the ability for the profile to be overridden on a per-container basis. This profile unfortunately prevents some of the mount() system calls that Singularity uses to set up user-mode containers from working inside the pod, even though these calls would be allowed for an unprivileged user under normal circumstances.

On the UCSC Kubernetes cluster, we configure our Ubuntu hosts with an alternative default AppArmor profile for Docker containers which allows these calls. Other solutions include turning off AppArmor on the host, configuring Kubernetes with a container runtime other than Docker, or using Kubernetes's AppArmor integration to apply a more permissive profile or the unconfined profile to pods that Toil launches.

Toil does not yet have a way to apply a container.apparmor.security.beta.kubernetes.io/runner-container: unconfined annotation to its pods, as described in the Kubernetes AppArmor documentation. This feature is tracked in issue #4331.

Running in AWS¶

Toil jobs can be run on a variety of cloud platforms. Of these, Amazon Web Services (AWS) is currently the best-supported solution. Toil provides the Toil Cluster Utilities to conveniently create AWS clusters, connect to the leader of the cluster, and then launch a workflow. The leader handles distributing the jobs over the worker nodes and autoscaling to optimize costs.

The Running a Workflow with Autoscaling section details how to create a cluster and run a workflow that will dynamically scale depending on the workflow's needs.

The Static Provisioning section explains how a static cluster (one that won't automatically change in size) can be created and provisioned (grown, shrunk, destroyed, etc.).

Preparing your AWS environment¶

To use Amazon Web Services (AWS) to run Toil or to just use S3 to host the files during the computation of a workflow, first set up and configure an account with AWS:

$ ssh-keygen -t rsa

~/.ssh/id_rsa

$ cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys

$ eval `ssh-agent -s`
$ ssh-add

$ chmod 400 id_rsa

https://us-west-1.console.aws.amazon.com/ec2/v2/home?region=us-west-1#KeyPairs:sort=keyName

https://console.aws.amazon.com/iam/home?region=us-west-1#/home

$ pip install awscli --upgrade --user

$ aws configure

" AWS Access Key ID [****************Q65Q]: "
" AWS Secret Access Key [****************G0ys]: "
" Default region name [us-west-1]: "
" Default output format [json]: "

$ virtualenv venv
$ source venv/bin/activate
$ pip install toil[all]==5.12.0

$ toil launch-cluster <cluster-name> \


      --clusterType kubernetes \


      --leaderNodeType t2.medium \


      --nodeTypes t2.medium -w 1 \


      --zone us-west-1a \


      --keyPairName id_rsa

AWS Job Store¶

Using the AWS job store is straightforward after you've finished Preparing your AWS environment; all you need to do is specify the prefix for the job store name.

To run the sort example sort example with the AWS job store you would type

$ python3 sort.py aws:us-west-2:my-aws-sort-jobstore

Toil Provisioner¶

The Toil provisioner is the component responsible for creating resources in Amazon's cloud. It is included in Toil alongside the [aws] extra and allows us to spin up a cluster.

Getting started with the provisioner is simple:

The Toil provisioner makes heavy use of the Toil Appliance, a Docker image that bundles Toil and all its requirements (e.g. Kubernetes). This makes deployment simple across platforms, and you can even simulate a cluster locally (see Developing with Docker for details).

For information on using the Toil Provisioner have a look at Running a Workflow with Autoscaling.

Details about Launching a Cluster in AWS¶

Using the provisioner to launch a Toil leader instance is simple using the launch-cluster command. For example, to launch a Kubernetes cluster named "my-cluster" with a t2.medium leader in the us-west-2a zone, run

(venv) $ toil launch-cluster my-cluster \


             --clusterType kubernetes \


             --leaderNodeType t2.medium \


             --nodeTypes t2.medium -w 1 \


             --zone us-west-2a \


             --keyPairName <AWS-key-pair-name>

The cluster name is used to uniquely identify your cluster and will be used to populate the instance's Name tag. Also, the Toil provisioner will automatically tag your cluster with an Owner tag that corresponds to your keypair name to facilitate cost tracking. In addition, the ToilNodeType tag can be used to filter "leader" vs. "worker" nodes in your cluster.

The leaderNodeType is an EC2 instance type. This only affects the leader node.

The --zone parameter specifies which EC2 availability zone to launch the cluster in. Alternatively, you can specify this option via the TOIL_AWS_ZONE environment variable. Note: the zone is different from an EC2 region. A region corresponds to a geographical area like us-west-2 (Oregon), and availability zones are partitions of this area like us-west-2a.

By default, Toil creates an IAM role for each cluster with sufficient permissions to perform cluster operations (e.g. full S3, EC2, and SDB access). If the default permissions are not sufficient for your use case (e.g. if you need access to ECR), you may create a custom IAM role with all necessary permissions and set the --awsEc2ProfileArn parameter when launching the cluster. Note that your custom role must at least have these permissions in order for the Toil cluster to function properly.

In addition, Toil creates a new security group with the same name as the cluster name with default rules (e.g. opens port 22 for SSH access). If you require additional security groups, you may use the --awsEc2ExtraSecurityGroupId parameter when launching the cluster. Note: Do not use the same name as the cluster name for the extra security groups as any security group matching the cluster name will be deleted once the cluster is destroyed.

For more information on options try:

(venv) $ toil launch-cluster --help

Static Provisioning¶

Toil can be used to manage a cluster in the cloud by using the Toil Cluster Utilities. The cluster utilities also make it easy to run a toil workflow directly on this cluster. We call this static provisioning because the size of the cluster does not change. This is in contrast with Running a Workflow with Autoscaling.

To launch worker nodes alongside the leader we use the -w option:

(venv) $ toil launch-cluster my-cluster \


             --clusterType kubernetes \


             --leaderNodeType t2.small -z us-west-2a \


             --keyPairName <AWS-key-pair-name> \


             --nodeTypes m3.large,t2.micro -w 1,4 \


             --zone us-west-2a

This will spin up a leader node of type t2.small with five additional workers --- one m3.large instance and four t2.micro.

Currently static provisioning is only possible during the cluster's creation. The ability to add new nodes and remove existing nodes via the native provisioner is in development. Of course the cluster can always be deleted with the Destroy-Cluster Command utility.

Uploading Workflows¶

Now that our cluster is launched, we use the Rsync-Cluster Command utility to copy the workflow to the leader. For a simple workflow in a single file this might look like

(venv) $ toil rsync-cluster -z us-west-2a my-cluster toil-workflow.py :/

NOTE:

Running a Workflow with Autoscaling¶

Toil can create an autoscaling Kubernetes cluster for you using the AWS provisioner. Autoscaling is a feature of running Toil in a cloud whereby additional cloud instances are launched as needed to run the workflow.

NOTE:

To set up a Kubernetes cluster, simply use the --clusterType=kubernetes command line option to toil launch-cluster. To make it autoscale, specify a range of possible node counts for a node type (such as -w 1-4). The cluster will automatically add and remove nodes, within that range, depending on how many seem to be needed to run the jobs submitted to the cluster.

For example, to launch a Toil cluster with a Kubernetes scheduler, run:

(venv) $ toil launch-cluster <cluster-name> \


        --provisioner=aws \


        --clusterType kubernetes \


        --zone us-west-2a \


        --keyPairName <AWS-key-pair-name> \


        --leaderNodeType t2.medium \


        --leaderStorage 50 \


        --nodeTypes t2.medium -w 1-4 \


        --nodeStorage 20 \


        --logDebug

Behind the scenes, Toil installs kubeadm and configures the kubelet on the Toil leader and all worker nodes. This Toil cluster can then schedule jobs using Kubernetes.

NOTE:

As a demonstration, we will use sort.py again, but run it on a Toil cluster with Kubernetes. First, download this file and put it to the current working directory.

We then need to copy over the workflow file and SSH into the cluster:

(venv) $ toil rsync-cluster -z us-west-2a <cluster-name> sort.py :/root
(venv) $ toil ssh-cluster -z us-west-2a <cluster-name>

Remember to replace <cluster-name> with your actual cluster name, and feel free to use your own cluster configuration and/or workflow files. For more information on this step, see the corresponding section of the Static Provisioning tutorial.

IMPORTANT:

Now that we are inside the cluster, a Kubernetes environment should already be configured and running. To verify this, simply run:

$ kubectl get nodes

You should see a leader node with the Ready status. Depending on the number of worker nodes you set to create upfront, you should also see them displayed here.

Additionally, you can also verify that the metrics server is running:

$ kubectl get --raw "/apis/metrics.k8s.io/v1beta1/nodes"

If there is a JSON response (similar to the output below), and you are not seeing any errors, that means the metrics server is set up and running, and you are good to start running workflows.

{"kind":"NodeMetricsList","apiVersion":"metrics.k8s.io/v1beta1", ...}

NOTE:

Now we can run the workflow:

$ python3 sort.py \


        --batchSystem kubernetes \


        aws:<region>:<job-store-name>

Make sure to replace <region> and <job-store-name>. It is required to use a cloud-accessible job store like AWS or Google when using the Kubernetes batch system.

The sort workflow should start running on the Kubernetes cluster set up by Toil. This workflow would take a while to execute, so you could put the job in the background and monitor the Kubernetes cluster using kubectl. For example, you can check out the pods that are running:

$ kubectl get pods

You should see an output like:

NAME                                                      READY   STATUS              RESTARTS   AGE
root-toil-a864e1b0-2e1f-48db-953c-038e5ad293c7-11-4cwdl   0/1     ContainerCreating   0          85s
root-toil-a864e1b0-2e1f-48db-953c-038e5ad293c7-14-5dqtk   0/1     Completed           0          18s
root-toil-a864e1b0-2e1f-48db-953c-038e5ad293c7-7-gkwc9    0/1     ContainerCreating   0          107s
root-toil-a864e1b0-2e1f-48db-953c-038e5ad293c7-9-t7vsb    1/1     Running             0          96s

If a pod failed for whatever reason or if you want to make sure a pod isn't stuck, you can use kubectl describe pod <pod-name> or kubectl logs <pod-name> to inspect the pod.

If everything is successful, you should be able to see an output file from the sort workflow:

$ head sortedFile.txt

You can now run your own workflows!

Preemptibility¶

Toil can run on a heterogeneous cluster of both preemptible and non-preemptible nodes. Being a preemptible node simply means that the node may be shut down at any time, while jobs are running. These jobs can then be restarted later somewhere else.

A node type can be specified as preemptible by adding a spot bid in dollars, after a colon, to its entry in the list of node types provided with the --nodeTypes flag. If spot instance prices rise above your bid, the preemptible nodes will be shut down.

For example, this cluster will have both preemptible and non-preemptible nodes:

(venv) $ toil launch-cluster <cluster-name> \


        --provisioner=aws \


        --clusterType kubernetes \


        --zone us-west-2a \


        --keyPairName <AWS-key-pair-name> \


        --leaderNodeType t2.medium \


        --leaderStorage 50 \


        --nodeTypes t2.medium -w 1-4 \


        --nodeTypes t2.large:0.20 -w 1-4 \


        --nodeStorage 20 \


        --logDebug

Individual jobs can explicitly specify whether they should be run on preemptible nodes via the boolean preemptible resource requirement in Toil's Python API. In CWL, this is exposed as a hint UsePreemptible in the http://arvados.org/cwl# namespace (usually imported as arv). In WDL, this is exposed as a runtime attribute preemptible as recognized by Cromwell. Toil's Kubernetes batch system will prefer to schedule preemptible jobs on preemptible nodes.

If a job is not specified to be preemptible, the job will not run on preemptible nodes even if preemptible nodes are available, unless the workflow is run with the --defaultPreemptible flag. The --defaultPreemptible flag will allow jobs without an explicit preemptible requirement to run on preemptible machines. For example:

$ python3 /root/sort.py aws:us-west-2:<my-jobstore-name> \


      --batchSystem kubernetes \


      --defaultPreemptible

Using MinIO and S3-Compatible object stores¶

Toil can be configured to access files stored in an S3-compatible object store such as MinIO. The following environment variables can be used to configure the S3 connection used:

Examples:

TOIL_S3_HOST=127.0.0.1
TOIL_S3_PORT=9010
TOIL_S3_USE_SSL=False

In-Workflow Autoscaling with Mesos¶

Instead of the normal Kubernetes-based autoscaling, you can also use Toil's old Mesos-based autoscaling method, where the scaling logic runs inside the Toil workflow. With this approach, a Toil cluster can only run one workflow at a time. This method also does not work on the ARM architecture.

In this mode, the --preemptibleCompensation flag can be used to handle cases where preemptible nodes may not be available but are required for your workflow. With this flag enabled, the autoscaler will attempt to compensate for a shortage of preemptible nodes of a certain type by creating non-preemptible nodes of that type, if non-preemptible nodes of that type were specified in --nodeTypes.

NOTE:

(venv) $ toil launch-cluster <cluster-name> \


             --clusterType mesos \


             --keyPairName <AWS-key-pair-name> \


             --leaderNodeType t2.medium \


             --zone us-west-2a

(venv) $ toil rsync-cluster -z us-west-2a <cluster-name> sort.py :/root

(venv) $ toil ssh-cluster -z us-west-2a <cluster-name>

$ python3 /root/sort.py aws:us-west-2:<my-jobstore-name> \


      --provisioner aws \


      --nodeTypes c3.large \


      --maxNodes 2 \


      --batchSystem mesos

NOTE:

$ head fileToSort.txt

$ head sortedFile.txt

Dashboard¶

Toil provides a dashboard for viewing the RAM and CPU usage of each node, the number of issued jobs of each type, the number of failed jobs, and the size of the jobs queue. To launch this dashboard for a Toil workflow, pass the --metrics flag on the workflow's command line. The dashboard can then be viewed in your browser at localhost:3000 while connected to the leader node through toil ssh-cluster:

To change the default port number, you can use the --grafana_port argument:

(venv) $ toil ssh-cluster -z us-west-2a --grafana_port 8000 <cluster-name>

On AWS, the dashboard keeps track of every node in the cluster to monitor CPU and RAM usage, but it can also be used while running a workflow on a single machine. The dashboard uses Grafana as the front end for displaying real-time plots, and Prometheus for tracking metrics exported by toil: [image]

In order to use the dashboard for a non-released toil version, you will have to build the containers locally with make docker, since the prometheus, grafana, and mtail containers used in the dashboard are tied to a specific toil version.

Running in Google Compute Engine (GCE)¶

Toil supports a provisioner with Google, and a Google Job Store. To get started, follow instructions for Preparing your Google environment.

Preparing your Google environment¶

Toil supports using the Google Cloud Platform. Setting this up is easy!

$ ssh-keygen -t rsa -f ~/.ssh/id_rsa -C [USERNAME]

$ chmod 400 ~/.ssh/id_rsa ~/.ssh/id_rsa.pub

For more details look at Google's instructions for adding SSH keys.

Google Job Store¶

To use the Google Job Store you will need to set the GOOGLE_APPLICATION_CREDENTIALS environment variable by following Google's instructions.

Then to run the sort example with the Google job store you would type

$ python3 sort.py google:my-project-id:my-google-sort-jobstore

Running a Workflow with Autoscaling¶

WARNING:

The steps to run a GCE workflow are similar to those of AWS (Running a Workflow with Autoscaling), except you will need to explicitly specify the --provisioner gce option which otherwise defaults to aws.

(venv) $ toil launch-cluster <CLUSTER-NAME> \


             --provisioner gce \


             --leaderNodeType n1-standard-1 \


             --keyPairName <SSH-KEYNAME> \


             --zone us-west1-a

(venv) $ toil rsync-cluster --provisioner gce <CLUSTER-NAME> sort.py :/root
(venv) $ toil ssh-cluster --provisioner gce <CLUSTER-NAME>

$ python3 /root/sort.py  google:<PROJECT-ID>:<JOBSTORE-NAME> \


      --provisioner gce \


      --batchSystem mesos \


      --nodeTypes n1-standard-2 \


      --maxNodes 2

$ exit  # this exits the ssh from the leader node
(venv) $ toil destroy-cluster --provisioner gce <CLUSTER-NAME>

Running on Slurm¶

When running Toil workflows on Slurm, you usually want to run the workflow itself from the head node. Toil will take care of running all the required sbatch commands for you. You probably do not want to submit the Toil workflow as a Slurm job with sbatch (although you can if you have a large number of workflows to run). You also probably do not want to manually allocate resources with sallocate.

To run a Toil workflow on Slurm, include --batchSystem slurm in your command line arguments. Generally Slurm clusters have shared filesystems, meaning the file job store would be appropriate. You want to make sure to use a job store location that is shared across your Slurm cluster. Additionally, you will likely want to provide another shared directory with the --batchLogsDir option, to allow the Slurm job logs to be retrieved by Toil in case something goes wrong with a job.

For example, to run the sort example sort example on Slurm, assuming you are currently in a shared directory, you would type, on the cluster head node:

$ mkdir -p logs
$ python3 sort.py ./store --batchSystem slurm --batchLogsDir ./logs

Slurm Tips¶

$ echo 'export SINGULARITY_CACHEDIR="${HOME}/.singularity/cache"' >>~/.bashrc
$ echo 'export MINIWDL__SINGULARITY__IMAGE_CACHE="${HOME}/.cache/miniwdl"' >>~/.bashrc

Standard Output/Error from Batch System Jobs¶

Standard output and error from batch system jobs (except for the Mesos batch system) are redirected to files in the toil-<workflowID> directory created within the temporary directory specified by the --workDir option; see Commandline Options. Each file is named as follows: toil_job_<Toil job ID>_batch_<name of batch system>_<job ID from batch system>_<file description>.log, where <file description> is std_output for standard output, and std_error for standard error. HTCondor will also write job event log files with <file description> = job_events.

If capturing standard output and error is desired, --workDir will generally need to be on a shared file system; otherwise if these are written to local temporary directories on each node (e.g. /tmp) Toil will not be able to retrieve them. Alternatively, the --noStdOutErr option forces Toil to discard all standard output and error from batch system jobs.

Preparing your WES environment¶

The WES server requires Celery to distribute and execute workflows. To set up Celery:

docker run -d --name wes-rabbitmq -p 5672:5672 rabbitmq:3.9.5

celery -A toil.server.celery_app worker --loglevel=INFO

Starting a WES server¶

To start a WES server on the default port 8080, run the Toil command:

$ toil server

The WES API will be hosted on the following URL:

http://localhost:8080/ga4gh/wes/v1

To use another port, e.g.: 3000, you can specify the --port argument:

$ toil server --port 3000

There are many other command line options. Help information can be found by using this command:

$ toil server --help

Below is a detailed summary of all server-specific options:

Running the Server with docker-compose¶

Instead of manually setting up the server components (toil server, RabbitMQ, and Celery), you can use the following docker-compose.yml file to orchestrate and link them together.

Make sure to change the credentials for basic authentication by updating the traefik.http.middlewares.auth.basicauth.users label. The passwords can be generated with tools like htpasswd like this. (Note that single $ signs need to be replaced with $$ in the yaml file).

When running on a different host other than localhost, make sure to change the Host to your tartget host in the traefik.http.routers.wes.rule and traefik.http.routers.wespublic.rule labels.

You can also change /tmp/toil-workflows if you want Toil workflows to live somewhere else, and create the directory before starting the server.

In order to run workflows that require Docker, the docker.sock socket must be mounted as volume for Celery. Additionally, the TOIL_WORKDIR directory (defaults to: /var/lib/toil) and /var/lib/cwl (if running CWL workflows with DockerRequirement) should exist on the host and also be mounted as volumes.

Also make sure to run it behind a firewall; it opens up the Toil server on port 8080 to anyone who connects.

# docker-compose.yml
version: "3.8"
services:


  rabbitmq:


    image: rabbitmq:3.9.5


    hostname: rabbitmq


  celery:


    image: ${TOIL_APPLIANCE_SELF}


    volumes:


      - /var/run/docker.sock:/var/run/docker.sock


      - /var/lib/docker:/var/lib/docker


      - /var/lib/toil:/var/lib/toil


      - /var/lib/cwl:/var/lib/cwl


      - /tmp/toil-workflows:/tmp/toil-workflows


    command: celery --broker=amqp://guest:guest@rabbitmq:5672// -A toil.server.celery_app worker --loglevel=INFO


    depends_on:


      - rabbitmq


  wes-server:


    image: ${TOIL_APPLIANCE_SELF}


    volumes:


      - /tmp/toil-workflows:/tmp/toil-workflows


    environment:


      - TOIL_WES_BROKER_URL=amqp://guest:guest@rabbitmq:5672//


    command: toil server --host 0.0.0.0 --port 8000 --work_dir /tmp/toil-workflows


    expose:


      - 8000


    labels:


      - "traefik.enable=true"


      - "traefik.http.routers.wes.rule=Host(`localhost`)"


      - "traefik.http.routers.wes.entrypoints=web"


      - "traefik.http.routers.wes.middlewares=auth"


      - "traefik.http.middlewares.auth.basicauth.users=test:$$2y$$12$$ci.4U63YX83CwkyUrjqxAucnmi2xXOIlEF6T/KdP9824f1Rf1iyNG"


      - "traefik.http.routers.wespublic.rule=Host(`localhost`) && Path(`/ga4gh/wes/v1/service-info`)"


    depends_on:


      - rabbitmq


      - celery


  traefik:


    image: traefik:v2.2


    command:


      - "--providers.docker"


      - "--providers.docker.exposedbydefault=false"


      - "--entrypoints.web.address=:8080"


    ports:


      - "8080:8080"


    volumes:


      - /var/run/docker.sock:/var/run/docker.sock

Further customization can also be made as needed. For example, if you have a domain, you can set up HTTPS with Let's Encrypt.

Once everything is configured, simply run docker-compose up to start the containers. Run docker-compose down to stop and remove all containers.

NOTE:

Running on a Toil cluster¶

To run the server on a Toil leader instance on EC2:

curl -L "https://github.com/docker/compose/releases/download/1.29.2/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
chmod +x /usr/local/bin/docker-compose
# check installation
docker-compose --version

WES API Endpoints¶

As defined by the GA4GH WES API specification, the following endpoints with base path ga4gh/wes/v1/ are supported by Toil:

When running the WES server with the docker-compose setup above, most endpoints (except GET /service-info) will be protected with basic authentication. Make sure to set the Authorization header with the correct credentials when submitting or retrieving a workflow.

Submitting a Workflow¶

Now that the WES API is up and running, we can submit and monitor workflows remotely using the WES API endpoints. A workflow can be submitted for execution using the POST /runs endpoint.

As a quick example, we can submit the example CWL workflow from Running a basic CWL workflow to our WES API:

# example.cwl
cwlVersion: v1.0
class: CommandLineTool
baseCommand: echo
stdout: output.txt
inputs:


 message:


   type: string


   inputBinding:


     position: 1
outputs:


 output:


   type: stdout

using cURL:

$ curl --location --request POST 'http://localhost:8080/ga4gh/wes/v1/runs' \


    --user test:test \


    --form 'workflow_url="example.cwl"' \


    --form 'workflow_type="cwl"' \


    --form 'workflow_type_version="v1.0"' \


    --form 'workflow_params="{\"message\": \"Hello world!\"}"' \


    --form 'workflow_attachment=@"./toil_test_files/example.cwl"'
{


  "run_id": "4deb8beb24894e9eb7c74b0f010305d1"
}

Note that the --user argument is used to attach the basic authentication credentials along with the request. Make sure to change test:test to the username and password you configured for your WES server. Alternatively, you can also set the Authorization header manually as "Authorization: Basic base64_encoded_auth".

If the workflow is submitted successfully, a JSON object containing a run_id will be returned. The run_id is a unique identifier of your requested workflow, which can be used to monitor or cancel the run.

There are a few required parameters that have to be set for all workflow submissions, which are the following:

Additionally, the following optional parameters are also available:

For more details about these parameters, refer to the Run Workflow section in the WES API spec.

Upload multiple files¶

Looking at the body of the request of the previous example, note that the workflow_url is a relative URL that refers to the example.cwl file uploaded from the local path ./toil_test_files/example.cwl.

To specify the file name (or subdirectory) of the remote destination file, set the filename field in the Content-Disposition header. You could also upload more than one file by providing the workflow_attachment parameter multiple times with different files.

This can be shown by the following example:

$ curl --location --request POST 'http://localhost:8080/ga4gh/wes/v1/runs' \


    --user test:test \


    --form 'workflow_url="example.cwl"' \


    --form 'workflow_type="cwl"' \


    --form 'workflow_type_version="v1.0"' \


    --form 'workflow_params="{\"message\": \"Hello world!\"}"' \


    --form 'workflow_attachment=@"./toil_test_files/example.cwl"' \


    --form 'workflow_attachment=@"./toil_test_files/2.fasta";filename=inputs/test.fasta' \


    --form 'workflow_attachment=@"./toil_test_files/2.fastq";filename=inputs/test.fastq'

On the server, the execution directory would have the following structure from the above request:

execution/
├── example.cwl
├── inputs
│     ├── test.fasta
|     └── test.fastq
└── wes_inputs.json

Specify Toil options¶

To pass Toil-specific parameters to the workflow, you can include the workflow_engine_parameters parameter along with your request.

For example, to set the logging level to INFO, and change the working directory of the workflow, simply include the following as workflow_engine_parameters:

{"--logLevel": "INFO", "--workDir": "/tmp/"}

These options would be appended at the end of existing parameters during command construction, which would override the default parameters if provided. (Default parameters that can be passed multiple times would not be overridden).

Monitoring a Workflow¶

With the run_id returned when submitting the workflow, we can check the status or get the full logs of the workflow run.

Checking the state¶

The GET /runs/{run_id}/status endpoint can be used to get a simple result with the overall state of your run:

$ curl --user test:test http://localhost:8080/ga4gh/wes/v1/runs/4deb8beb24894e9eb7c74b0f010305d1/status
{


  "run_id": "4deb8beb24894e9eb7c74b0f010305d1",


  "state": "RUNNING"
}

The possible states here are: QUEUED, INITIALIZING, RUNNING, COMPLETE, EXECUTOR_ERROR, SYSTEM_ERROR, CANCELING, and CANCELED.

Getting the full logs¶

To get the detailed information about a workflow run, use the GET /runs/{run_id} endpoint:

$ curl --user test:test http://localhost:8080/ga4gh/wes/v1/runs/4deb8beb24894e9eb7c74b0f010305d1
{


  "run_id": "4deb8beb24894e9eb7c74b0f010305d1",


  "request": {?


    "workflow_attachment": [


      "example.cwl"


    ],


    "workflow_url": "example.cwl",


    "workflow_type": "cwl",


    "workflow_type_version": "v1.0",


    "workflow_params": {


      "message": "Hello world!"


    }


  },


  "state": "RUNNING",


  "run_log": {


    "cmd": [


      "toil-cwl-runner --outdir=/home/toil/workflows/4deb8beb24894e9eb7c74b0f010305d1/outputs --jobStore=file:/home/toil/workflows/4deb8beb24894e9eb7c74b0f010305d1/toil_job_store /home/toil/workflows/4deb8beb24894e9eb7c74b0f010305d1/execution/example.cwl /home/workflows/4deb8beb24894e9eb7c74b0f010305d1/execution/wes_inputs.json"


    ],


    "start_time": "2021-08-30T17:35:50Z",


    "end_time": null,


    "stdout": null,


    "stderr": null,


    "exit_code": null


  },


  "task_logs": [],


  "outputs": {}
}

Canceling a run¶

To cancel a workflow run, use the POST /runs/{run_id}/cancel endpoint:

$ curl --location --request POST 'http://localhost:8080/ga4gh/wes/v1/runs/4deb8beb24894e9eb7c74b0f010305d1/cancel' \


      --user test:test
{


  "run_id": "4deb8beb24894e9eb7c74b0f010305d1"
}

Scripting Quick Start¶

To begin, consider this short Toil Python workflow which illustrates defining a workflow:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
def helloWorld(message):


    return f"Hello, world!, here's a message: {message}"
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_quickstart")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "OFF"


    options.clean = "always"


    hello_job = Job.wrapFn(helloWorld, "Woot")


    with Toil(options) as toil:


        print(toil.start(hello_job))  # prints "Hello, world!, ..."

The workflow consists of a single job. The resource requirements for that job are (optionally) specified by keyword arguments (memory, cores, disk). The workflow is run using toil.job.Job.Runner.getDefaultOptions(). Below we explain the components of this code in detail.

Job Basics¶

The atomic unit of work in a Toil workflow is a Job. User code extends this base class, or uses helper methods like toil.job.Job.addChildJobFn(), to define units of work. For example, here is a more long-winded class-based version of the job in the quick start example:

from toil.job import Job
class HelloWorld(Job):


    def __init__(self, message):


        Job.__init__(self,  memory="2G", cores=2, disk="3G")


        self.message = message


    def run(self, fileStore):


        return f"Hello, world! Here's a message: {self.message}"

In the example a class, HelloWorld, is defined. The constructor requests 2 gigabytes of memory, 2 cores and 3 gigabytes of local disk to complete the work.

The toil.job.Job.run() method is the function the user overrides to get work done. Here it just returns a message.

It is also possible to log a message using toil.job.Job.log(), which will be registered in the log output of the leader process of the workflow:

...


    def run(self, fileStore):


        self.log(f"Hello, world! Here's a message: {self.message}")

Invoking a Workflow¶

We can add to the previous example to turn it into a complete workflow by adding the necessary function calls to create an instance of HelloWorld and to run this as a workflow containing a single job. For example:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
class HelloWorld(Job):


    def __init__(self, message):


        Job.__init__(self)


        self.message = message


    def run(self, fileStore):


        return f"Hello, world!, here's a message: {self.message}"
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_invokeworkflow")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "OFF"


    options.clean = "always"


    hello_job = HelloWorld("Woot")


    with Toil(options) as toil:


        print(toil.start(hello_job))

NOTE:

This uses the toil.common.Toil class, which is used to run and resume Toil workflows. It is used as a context manager and allows for preliminary setup, such as staging of files into the job store on the leader node. An instance of the class is initialized by specifying an options object. The actual workflow is then invoked by calling the toil.common.Toil.start() method, passing the root job of the workflow, or, if a workflow is being restarted, toil.common.Toil.restart() should be used. Note that the context manager should have explicit if else branches addressing restart and non restart cases. The boolean value for these if else blocks is toil.options.restart.

For example:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
class HelloWorld(Job):


    def __init__(self, message):


        Job.__init__(self)


        self.message = message


    def run(self, fileStore):


        return f"Hello, world!, I have a message: {self.message}"
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_invokeworkflow2")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        if not toil.options.restart:


            job = HelloWorld("Woot!")


            output = toil.start(job)


        else:


            output = toil.restart()


    print(output)

The call to toil.job.Job.Runner.getDefaultOptions() creates a set of default options for the workflow. The only argument is a description of how to store the workflow's state in what we call a job-store. Here the job-store is contained in a directory within the current working directory called "toilWorkflowRun". Alternatively this string can encode other ways to store the necessary state, e.g. an S3 bucket object store location. By default the job-store is deleted if the workflow completes successfully.

The workflow is executed in the final line, which creates an instance of HelloWorld and runs it as a workflow. Note all Toil workflows start from a single starting job, referred to as the root job. The return value of the root job is returned as the result of the completed workflow (see promises below to see how this is a useful feature!).

Specifying Commandline Arguments¶

To allow command line control of the options we can use the toil.job.Job.Runner.getDefaultArgumentParser() method to create a argparse.ArgumentParser object which can be used to parse command line options for a Toil Python workflow. For example:

from toil.common import Toil
from toil.job import Job
class HelloWorld(Job):


    def __init__(self, message):


        Job.__init__(self)


        self.message = message


    def run(self, fileStore):


        return "Hello, world!, here's a message: %s" % self.message
if __name__ == "__main__":


    parser = Job.Runner.getDefaultArgumentParser()


    options = parser.parse_args()


    options.logLevel = "OFF"


    options.clean = "always"


    hello_job = HelloWorld("Woot")


    with Toil(options) as toil:


        print(toil.start(hello_job))

This creates a fully fledged Toil Python workflow with all the options Toil exposes as command line arguments. Running this program with --help will print the full list of options.

Alternatively an existing argparse.ArgumentParser object can have Toil command line options added to it with the toil.job.Job.Runner.addToilOptions() method.

Resuming a Workflow¶

In the event that a workflow fails, either because of programmatic error within the jobs being run, or because of node failure, the workflow can be resumed. Workflows can only not be reliably resumed if the job-store itself becomes corrupt.

Critical to resumption is that jobs can be rerun, even if they have apparently completed successfully. Put succinctly, a user defined job should not corrupt its input arguments. That way, regardless of node, network or leader failure the job can be restarted and the workflow resumed.

To resume a workflow specify the "restart" option in the options object passed to toil.common.Toil.start(). If node failures are expected it can also be useful to use the integer "retryCount" option, which will attempt to rerun a job retryCount number of times before marking it fully failed.

In the common scenario that a small subset of jobs fail (including retry attempts) within a workflow Toil will continue to run other jobs until it can do no more, at which point toil.common.Toil.start() will raise a toil.exceptions.FailedJobsException exception. Typically at this point the user can decide to fix the script and resume the workflow or delete the job-store manually and rerun the complete workflow.

Functions and Job Functions¶

Defining jobs by creating class definitions generally involves the boilerplate of creating a constructor. To avoid this the classes toil.job.FunctionWrappingJob and toil.job.JobFunctionWrappingTarget allow functions to be directly converted to jobs. For example, the quick start example (repeated here):

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
def helloWorld(message):


    return f"Hello, world!, here's a message: {message}"
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_quickstart")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "OFF"


    options.clean = "always"


    hello_job = Job.wrapFn(helloWorld, "Woot")


    with Toil(options) as toil:


        print(toil.start(hello_job))  # prints "Hello, world!, ..."

Is equivalent to the previous example, but using a function to define the job.

The function call:

Job.wrapFn(helloWorld, "Woot")

Creates the instance of the toil.job.FunctionWrappingTarget that wraps the function.

The keyword arguments memory, cores and disk allow resource requirements to be specified as before. Even if they are not included as keyword arguments within a function header they can be passed as arguments when wrapping a function as a job and will be used to specify resource requirements.

We can also use the function wrapping syntax to a job function, a function whose first argument is a reference to the wrapping job. Just like a self argument in a class, this allows access to the methods of the wrapping job, see toil.job.JobFunctionWrappingTarget. For example:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
def helloWorld(job, message):


    job.log(f"Hello world, I have a message: {message}")
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_jobfunctions")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    hello_job = Job.wrapJobFn(helloWorld, "Woot!")


    with Toil(options) as toil:


        toil.start(hello_job)

Here helloWorld() is a job function. It uses the toil.job.Job.log() to log a message that will be printed to the output console. Here the only subtle difference to note is the line:

hello_job = Job.wrapJobFn(helloWorld, "Woot")

Which uses the function toil.job.Job.wrapJobFn() to wrap the job function instead of toil.job.Job.wrapFn() which wraps a vanilla function.

Workflows with Multiple Jobs¶

A parent job can have child jobs and follow-on jobs. These relationships are specified by methods of the job class, e.g. toil.job.Job.addChild() and toil.job.Job.addFollowOn().

Considering a set of jobs the nodes in a job graph and the child and follow-on relationships the directed edges of the graph, we say that a job B that is on a directed path of child/follow-on edges from a job A in the job graph is a successor of A, similarly A is a predecessor of B.

A parent job's child jobs are run directly after the parent job has completed, and in parallel. The follow-on jobs of a job are run after its child jobs and their successors have completed. They are also run in parallel. Follow-ons allow the easy specification of cleanup tasks that happen after a set of parallel child tasks. The following shows a simple example that uses the earlier helloWorld() job function:

from toil.common import Toil
from toil.job import Job
def helloWorld(job, message):


    job.log(f"Hello world, I have a message: {message}")
if __name__ == "__main__":


    parser = Job.Runner.getDefaultArgumentParser()


    options = parser.parse_args()


    options.logLevel = "INFO"


    options.clean = "always"


    j1 = Job.wrapJobFn(helloWorld, "first")


    j2 = Job.wrapJobFn(helloWorld, "second or third")


    j3 = Job.wrapJobFn(helloWorld, "second or third")


    j4 = Job.wrapJobFn(helloWorld, "last")


    j1.addChild(j2)


    j1.addChild(j3)


    j1.addFollowOn(j4)


    with Toil(options) as toil:


        toil.start(j1)

In the example four jobs are created, first j1 is run, then j2 and j3 are run in parallel as children of j1, finally j4 is run as a follow-on of j1.

There are multiple short hand functions to achieve the same workflow, for example:

from toil.common import Toil
from toil.job import Job
def helloWorld(job, message):


    job.log(f"Hello world, I have a message: {message}")
if __name__ == "__main__":


    parser = Job.Runner.getDefaultArgumentParser()


    options = parser.parse_args()


    options.logLevel = "INFO"


    options.clean = "always"


    j1 = Job.wrapJobFn(helloWorld, "first")


    j2 = j1.addChildJobFn(helloWorld, "second or third")


    j3 = j1.addChildJobFn(helloWorld, "second or third")


    j4 = j1.addFollowOnJobFn(helloWorld, "last")


    with Toil(options) as toil:


        toil.start(j1)

Equivalently defines the workflow, where the functions toil.job.Job.addChildJobFn() and toil.job.Job.addFollowOnJobFn() are used to create job functions as children or follow-ons of an earlier job.

Jobs graphs are not limited to trees, and can express arbitrary directed acyclic graphs. For a precise definition of legal graphs see toil.job.Job.checkJobGraphForDeadlocks(). The previous example could be specified as a DAG as follows:

from toil.common import Toil
from toil.job import Job
def helloWorld(job, message):


    job.log(f"Hello world, I have a message: {message}")
if __name__ == "__main__":


    parser = Job.Runner.getDefaultArgumentParser()


    options = parser.parse_args()


    options.logLevel = "INFO"


    options.clean = "always"


    j1 = Job.wrapJobFn(helloWorld, "first")


    j2 = j1.addChildJobFn(helloWorld, "second or third")


    j3 = j1.addChildJobFn(helloWorld, "second or third")


    j4 = j2.addChildJobFn(helloWorld, "last")


    j3.addChild(j4)


    with Toil(options) as toil:


        toil.start(j1)

Note the use of an extra child edge to make j4 a child of both j2 and j3.

Dynamic Job Creation¶

The previous examples show a workflow being defined outside of a job. However, Toil also allows jobs to be created dynamically within jobs. For example:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
def binaryStringFn(job, depth, message=""):


    if depth > 0:


        job.addChildJobFn(binaryStringFn, depth-1, message + "0")


        job.addChildJobFn(binaryStringFn, depth-1, message + "1")


    else:


        job.log(f"Binary string: {message}")
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_dynamic")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        toil.start(Job.wrapJobFn(binaryStringFn, depth=5))

The job function binaryStringFn logs all possible binary strings of length n (here n=5), creating a total of 2^(n+2) - 1 jobs dynamically and recursively. Static and dynamic creation of jobs can be mixed in a Toil workflow, with jobs defined within a job or job function being created at run time.

Promises¶

The previous example of dynamic job creation shows variables from a parent job being passed to a child job. Such forward variable passing is naturally specified by recursive invocation of successor jobs within parent jobs. This can also be achieved statically by passing around references to the return variables of jobs. In Toil this is achieved with promises, as illustrated in the following example:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
def fn(job, i):


    job.log("i is: %s" % i, level=100)


    return i + 1
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_promises")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    j1 = Job.wrapJobFn(fn, 1)


    j2 = j1.addChildJobFn(fn, j1.rv())


    j3 = j1.addFollowOnJobFn(fn, j2.rv())


    with Toil(options) as toil:


        toil.start(j1)

Running this workflow results in three log messages from the jobs: i is 1 from j1, i is 2 from j2 and i is 3 from j3.

The return value from the first job is promised to the second job by the call to toil.job.Job.rv() in the following line:

j2 = j1.addChildFn(fn, j1.rv())

The value of j1.rv() is a promise, rather than the actual return value of the function, because j1 for the given input has at that point not been evaluated. A promise (toil.job.Promise) is essentially a pointer to for the return value that is replaced by the actual return value once it has been evaluated. Therefore, when j2 is run the promise becomes 2.

Promises also support indexing of return values:

def parent(job):


    indexable = Job.wrapJobFn(fn)


    job.addChild(indexable)


    job.addFollowOnFn(raiseWrap, indexable.rv(2))
def raiseWrap(arg):


    raise RuntimeError(arg) # raises "2"
def fn(job):


    return (0, 1, 2, 3)

Promises can be quite useful. For example, we can combine dynamic job creation with promises to achieve a job creation process that mimics the functional patterns possible in many programming languages:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
def binaryStrings(job, depth, message=""):


    if depth > 0:


        s = [job.addChildJobFn(binaryStrings, depth - 1, message + "0").rv(),


             job.addChildJobFn(binaryStrings, depth - 1, message + "1").rv()]


        return job.addFollowOnFn(merge, s).rv()


    return [message]
def merge(strings):


    return strings[0] + strings[1]
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_promises2")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.loglevel = "OFF"


    options.clean = "always"


    with Toil(options) as toil:


        print(toil.start(Job.wrapJobFn(binaryStrings, depth=5)))

The return value l of the workflow is a list of all binary strings of length 10, computed recursively. Although a toy example, it demonstrates how closely Toil workflows can mimic typical programming patterns.

Promised Requirements¶

Promised requirements are a special case of Promises that allow a job's return value to be used as another job's resource requirements.

This is useful when, for example, a job's storage requirement is determined by a file staged to the job store by an earlier job:

import os
from toil.common import Toil
from toil.job import Job, PromisedRequirement
from toil.lib.io import mkdtemp
def parentJob(job):


    downloadJob = Job.wrapJobFn(stageFn, "file://" + os.path.realpath(__file__), cores=0.1, memory='32M', disk='1M')


    job.addChild(downloadJob)


    analysis = Job.wrapJobFn(analysisJob,


                             fileStoreID=downloadJob.rv(0),


                             disk=PromisedRequirement(downloadJob.rv(1)))


    job.addFollowOn(analysis)
def stageFn(job, url):


    importedFile = job.fileStore.import_file(url)


    return importedFile, importedFile.size
def analysisJob(job, fileStoreID):


    # now do some analysis on the file


    pass
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_requirements")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        toil.start(Job.wrapJobFn(parentJob))

Note that this also makes use of the size attribute of the FileID object. This promised requirements mechanism can also be used in combination with an aggregator for multiple jobs' output values:

def parentJob(job):


    aggregator = []


    for fileNum in range(0, 10):


        downloadJob = Job.wrapJobFn(stageFn, "file://" + os.path.realpath(__file__), cores=0.1, memory='32M', disk='1M')


        job.addChild(downloadJob)


        aggregator.append(downloadJob)


    analysis = Job.wrapJobFn(analysisJob,


                             fileStoreID=downloadJob.rv(0),


                             disk=PromisedRequirement(lambda xs: sum(xs), [j.rv(1) for j in aggregator]))


    job.addFollowOn(analysis)

FileID¶

The toil.fileStore.FileID class is a small wrapper around Python's builtin string class. It is used to represent a file's ID in the file store, and has a size attribute that is the file's size in bytes. This object is returned by importFile and writeGlobalFile.

Managing files within a workflow¶

It is frequently the case that a workflow will want to create files, both persistent and temporary, during its run. The toil.fileStores.abstractFileStore.AbstractFileStore class is used by jobs to manage these files in a manner that guarantees cleanup and resumption on failure.

The toil.job.Job.run() method has a file store instance as an argument. The following example shows how this can be used to create temporary files that persist for the length of the job, be placed in a specified local disk of the node and that will be cleaned up, regardless of failure, when the job finishes:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
class LocalFileStoreJob(Job):


    def run(self, fileStore):


        # self.tempDir will always contain the name of a directory within the allocated disk space reserved for the job


        scratchDir = self.tempDir


        # Similarly create a temporary file.


        scratchFile = fileStore.getLocalTempFile()
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_managing")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    # Create an instance of FooJob which will have at least 2 gigabytes of storage space.


    j = LocalFileStoreJob(disk="2G")


    # Run the workflow


    with Toil(options) as toil:


        toil.start(j)

Job functions can also access the file store for the job. The equivalent of the LocalFileStoreJob class is

def localFileStoreJobFn(job):


    scratchDir = job.tempDir


    scratchFile = job.fileStore.getLocalTempFile()

Note that the fileStore attribute is accessed as an attribute of the job argument.

In addition to temporary files that exist for the duration of a job, the file store allows the creation of files in a global store, which persists during the workflow and are globally accessible (hence the name) between jobs. For example:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
def globalFileStoreJobFn(job):


    job.log("The following example exercises all the methods provided "


            "by the toil.fileStores.abstractFileStore.AbstractFileStore class")


    # Create a local temporary file.


    scratchFile = job.fileStore.getLocalTempFile()


    # Write something in the scratch file.


    with open(scratchFile, 'w') as fH:


        fH.write("What a tangled web we weave")


    # Write a copy of the file into the file-store; fileID is the key that can be used to retrieve the file.


    # This write is asynchronous by default


    fileID = job.fileStore.writeGlobalFile(scratchFile)


    # Write another file using a stream; fileID2 is the


    # key for this second file.


    with job.fileStore.writeGlobalFileStream(cleanup=True) as (fH, fileID2):


        fH.write(b"Out brief candle")


    # Now read the first file; scratchFile2 is a local copy of the file that is read-only by default.


    scratchFile2 = job.fileStore.readGlobalFile(fileID)


    # Read the second file to a desired location: scratchFile3.


    scratchFile3 = os.path.join(job.tempDir, "foo.txt")


    job.fileStore.readGlobalFile(fileID2, userPath=scratchFile3)


    # Read the second file again using a stream.


    with job.fileStore.readGlobalFileStream(fileID2) as fH:


        print(fH.read())  # This prints "Out brief candle"


    # Delete the first file from the global file-store.


    job.fileStore.deleteGlobalFile(fileID)


    # It is unnecessary to delete the file keyed by fileID2 because we used the cleanup flag,


    # which removes the file after this job and all its successors have run (if the file still exists)
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_managing2")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        toil.start(Job.wrapJobFn(globalFileStoreJobFn))

The example demonstrates the global read, write and delete functionality of the file-store, using both local copies of the files and streams to read and write the files. It covers all the methods provided by the file store interface.

What is obvious is that the file-store provides no functionality to update an existing "global" file, meaning that files are, barring deletion, immutable. Also worth noting is that there is no file system hierarchy for files in the global file store. These limitations allow us to fairly easily support different object stores and to use caching to limit the amount of network file transfer between jobs.

Staging of Files into the Job Store¶

External files can be imported into or exported out of the job store prior to running a workflow when the toil.common.Toil context manager is used on the leader. The context manager provides methods toil.common.Toil.importFile(), and toil.common.Toil.exportFile() for this purpose. The destination and source locations of such files are described with URLs passed to the two methods. Local files can be imported and exported as relative paths, and should be relative to the directory where the toil workflow is initially run from.

Using absolute paths and appropriate schema where possible (prefixing with "file://" or "s3:/" for example), make imports and exports less ambiguous and is recommended.

A list of the currently supported URLs can be found at toil.jobStores.abstractJobStore.AbstractJobStore.importFile(). To import an external file into the job store as a shared file, pass the optional sharedFileName parameter to that method.

If a workflow fails for any reason an imported file acts as any other file in the job store. If the workflow was configured such that it not be cleaned up on a failed run, the file will persist in the job store and needs not be staged again when the workflow is resumed.

Example:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
class HelloWorld(Job):


    def __init__(self, id):


        Job.__init__(self)


        self.inputFileID = id


    def run(self, fileStore):


        with fileStore.readGlobalFileStream(self.inputFileID, encoding='utf-8') as fi:


            with fileStore.writeGlobalFileStream(encoding='utf-8') as (fo, outputFileID):


                fo.write(fi.read() + 'World!')


        return outputFileID
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_staging")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        if not toil.options.restart:


            ioFileDirectory = os.path.join(os.path.dirname(os.path.abspath(__file__)), "stagingExampleFiles")


            inputFileID = toil.importFile("file://" + os.path.abspath(os.path.join(ioFileDirectory, "in.txt")))


            outputFileID = toil.start(HelloWorld(inputFileID))


        else:


            outputFileID = toil.restart()


        toil.exportFile(outputFileID, "file://" + os.path.abspath(os.path.join(ioFileDirectory, "out.txt")))

Using Docker Containers in Toil¶

Docker containers are commonly used with Toil. The combination of Toil and Docker allows for pipelines to be fully portable between any platform that has both Toil and Docker installed. Docker eliminates the need for the user to do any other tool installation or environment setup.

In order to use Docker containers with Toil, Docker must be installed on all workers of the cluster. Instructions for installing Docker can be found on the Docker website.

When using Toil-based autoscaling, Docker will be automatically set up on the cluster's worker nodes, so no additional installation steps are necessary. Further information on using Toil-based autoscaling can be found in the Running a Workflow with Autoscaling documentation.

In order to use docker containers in a Toil workflow, the container can be built locally or downloaded in real time from an online docker repository like Quay. If the container is not in a repository, the container's layers must be accessible on each node of the cluster.

When invoking docker containers from within a Toil workflow, it is strongly recommended that you use dockerCall(), a toil job function provided in toil.lib.docker. dockerCall leverages docker's own python API, and provides container cleanup on job failure. When docker containers are run without this feature, failed jobs can result in resource leaks. Docker's API can be found at docker-py.

In order to use dockerCall, your installation of Docker must be set up to run without sudo. Instructions for setting this up can be found here.

An example of a basic dockerCall is below:

dockerCall(job=job,


            tool='quay.io/ucsc_cgl/bwa',


            workDir=job.tempDir,


            parameters=['index', '/data/reference.fa'])

Note the assumption that reference.fa file is located in /data. This is Toil's standard convention as a mount location to reduce boilerplate when calling dockerCall. Users can choose their own mount locations by supplying a volumes kwarg to dockerCall, such as: volumes={working_dir: {'bind': '/data', 'mode': 'rw'}}, where working_dir is an absolute path on the user's filesystem.

dockerCall can also be added to workflows like any other job function:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.docker import apiDockerCall
from toil.lib.io import mkdtemp
align = Job.wrapJobFn(apiDockerCall,


                      image='ubuntu',


                      working_dir=os.getcwd(),


                      parameters=['ls', '-lha'])
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_docker")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        toil.start(align)

cgl-docker-lib contains dockerCall-compatible Dockerized tools that are commonly used in bioinformatics analysis.

The documentation provides guidelines for developing your own Docker containers that can be used with Toil and dockerCall. In order for a container to be compatible with dockerCall, it must have an ENTRYPOINT set to a wrapper script, as described in cgl-docker-lib containerization standards. This can be set by passing in the optional keyword argument, 'entrypoint'. Example:

dockerCall supports currently the 75 keyword arguments found in the python Docker API, under the 'run' command.

Services¶

It is sometimes desirable to run services, such as a database or server, concurrently with a workflow. The toil.job.Job.Service class provides a simple mechanism for spawning such a service within a Toil workflow, allowing precise specification of the start and end time of the service, and providing start and end methods to use for initialization and cleanup. The following simple, conceptual example illustrates how services work:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
class DemoService(Job.Service):


    def start(self, fileStore):


        # Start up a database/service here


        # Return a value that enables another process to connect to the database


        return "loginCredentials"


    def check(self):


        # A function that if it returns False causes the service to quit


        # If it raises an exception the service is killed and an error is reported


        return True


    def stop(self, fileStore):


        # Cleanup the database here


        pass
j = Job()
s = DemoService()
loginCredentialsPromise = j.addService(s)
def dbFn(loginCredentials):


    # Use the login credentials returned from the service's start method to connect to the service


    pass
j.addChildFn(dbFn, loginCredentialsPromise)
if __name__ == "__main__":


    jobstore: str = mkdtemp("tutorial_services")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        toil.start(j)

In this example the DemoService starts a database in the start method, returning an object from the start method indicating how a client job would access the database. The service's stop method cleans up the database, while the service's check method is polled periodically to check the service is alive.

A DemoService instance is added as a service of the root job j, with resource requirements specified. The return value from toil.job.Job.addService() is a promise to the return value of the service's start method. When the promised is fulfilled it will represent how to connect to the database. The promise is passed to a child job of j, which uses it to make a database connection. The services of a job are started before any of its successors have been run and stopped after all the successors of the job have completed successfully.

Multiple services can be created per job, all run in parallel. Additionally, services can define sub-services using toil.job.Job.Service.addChild(). This allows complex networks of services to be created, e.g. Apache Spark clusters, within a workflow.

Checkpoints¶

Services complicate resuming a workflow after failure, because they can create complex dependencies between jobs. For example, consider a service that provides a database that multiple jobs update. If the database service fails and loses state, it is not clear that just restarting the service will allow the workflow to be resumed, because jobs that created that state may have already finished. To get around this problem Toil supports checkpoint jobs, specified as the boolean keyword argument checkpoint to a job or wrapped function, e.g.:

j = Job(checkpoint=True)

A checkpoint job is rerun if one or more of its successors fails its retry attempts, until it itself has exhausted its retry attempts. Upon restarting a checkpoint job all its existing successors are first deleted, and then the job is rerun to define new successors. By checkpointing a job that defines a service, upon failure of the service the database and the jobs that access the service can be redefined and rerun.

To make the implementation of checkpoint jobs simple, a job can only be a checkpoint if when first defined it has no successors, i.e. it can only define successors within its run method.

Encapsulation¶

Let A be a root job potentially with children and follow-ons. Without an encapsulated job the simplest way to specify a job B which runs after A and all its successors is to create a parent of A, call it Ap, and then make B a follow-on of Ap. e.g.:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
if __name__ == "__main__":


    # A is a job with children and follow-ons, for example:


    A = Job()


    A.addChild(Job())


    A.addFollowOn(Job())


    # B is a job which needs to run after A and its successors


    B = Job()


    # The way to do this without encapsulation is to make a parent of A, Ap, and make B a follow-on of Ap.


    Ap = Job()


    Ap.addChild(A)


    Ap.addFollowOn(B)


    jobstore: str = mkdtemp("tutorial_encapsulations")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        print(toil.start(Ap))

An encapsulated job E(A) of A saves making Ap, instead we can write:

import os
from toil.common import Toil
from toil.job import Job
from toil.lib.io import mkdtemp
if __name__ == "__main__":


    # A


    A = Job()


    A.addChild(Job())


    A.addFollowOn(Job())


    # Encapsulate A


    A = A.encapsulate()


    # B is a job which needs to run after A and its successors


    B = Job()


    # With encapsulation A and its successor subgraph appear to be a single job, hence:


    A.addChild(B)


    jobstore: str = mkdtemp("tutorial_encapsulations2")


    os.rmdir(jobstore)


    options = Job.Runner.getDefaultOptions(jobstore)


    options.logLevel = "INFO"


    options.clean = "always"


    with Toil(options) as toil:


        print(toil.start(A))

Note the call to toil.job.Job.encapsulate() creates the toil.job.Job.EncapsulatedJob.

Depending on Toil¶

If you are packing your workflow(s) as a pip-installable distribution on PyPI, you might be tempted to declare Toil as a dependency in your setup.py, via the install_requires keyword argument to setup(). Unfortunately, this does not work, for two reasons: For one, Toil uses Setuptools' extra mechanism to manage its own optional dependencies. If you explicitly declared a dependency on Toil, you would have to hard-code a particular combination of extras (or no extras at all), robbing the user of the choice what Toil extras to install. Secondly, and more importantly, declaring a dependency on Toil would only lead to Toil being installed on the leader node of a cluster, but not the worker nodes. Auto-deployment does not work here because Toil cannot auto-deploy itself, the classic "Which came first, chicken or egg?" problem.

In other words, you shouldn't explicitly depend on Toil. Document the dependency instead (as in "This workflow needs Toil version X.Y.Z to be installed") and optionally add a version check to your setup.py. Refer to the check_version() function in the toil-lib project's setup.py for an example. Alternatively, you can also just depend on toil-lib and you'll get that check for free.

If your workflow depends on a dependency of Toil, consider not making that dependency explicit either. If you do, you risk a version conflict between your project and Toil. The pip utility may silently ignore that conflict, breaking either Toil or your workflow. It is safest to simply assume that Toil installs that dependency for you. The only downside is that you are locked into the exact version of that dependency that Toil declares. But such is life with Python, which, unlike Java, has no means of dependencies belonging to different software components within the same process, and whose favored software distribution utility is incapable of properly resolving overlapping dependencies and detecting conflicts.

Best Practices for Dockerizing Toil Workflows¶

Computational Genomics Lab's Dockstore based production system provides workflow authors a way to run Dockerized versions of their pipeline in an automated, scalable fashion. To be compatible with this system of a workflow should meet the following requirements. In addition to the Docker container, a common workflow language descriptor file is needed. For inputs:

For outputs:

FunctionWrappingJob¶

The subclass of Job for wrapping user functions.

JobFunctionWrappingJob¶

The subclass of FunctionWrappingJob for wrapping user job functions.

Job.wrapJobFn(myJob, memory='100k', disk='1M', cores=0.1)