wr as new Nextflow backend

[sb10]

New feature

I develop wr which is a workflow runner like Nextflow, but can also just be used as a backend scheduler. It can schedule to LSF and OpenStack right now.

The benefit to Nextflow users of going via wr instead of using Nextflow’s existing LSF or Kubernetes support is:

1. wr makes more efficient use of LSF: it can pick an appropriate queue, use job arrays, and “reuse” job slots. In a simple test I did, Nextflow using wr in LSF mode was 2 times faster than Nextflow using its own LSF scheduler.
2. wr’s OpenStack support is incredibly easy to use and set up (basically a single command to run), and provides auto scaling up and down. Kubernetes, by comparison, is really quite complex to get working on OpenStack, doesn’t auto scale, and wastes resources with multiple nodes needed even while no workflows are being operated on. I was able to get Nextflow to work with wr in OpenStack mode (but the shared disk requirement for Nextflow’s state remains a concern).

Usage scenario

Users with access to LSF or OpenStack clusters who want to run their Nextflow workflows efficiently and easily.

Suggest implementation

Since I don’t know Java well enough to understand how to implement this “correctly”, I wrote a simple bsub emulator in wr, which is what my tests so far have been based on. I submit the Nextflow command as a job to wr, turning on the bsub emulation, and configure Nextflow to use its existing LSF scheduler. While running under the emulation, Nextflow’s bsub calls actually call wr.

Of course the proper way to do this would be have Nextflow call wr directly (either the wr command line, or it’s REST API). The possibly tricky thing with regard to having it work in OpenStack mode is having it tell wr about OpenStack-specific things like what image to use, what hardware flavour to use, pass details on how to mount S3 etc. (the bsub emulation handles all of this).

Here's what I did for my LSF test...

echo_1000_sleep.nf:

#!/usr/bin/env nextflow

num = Channel.from(1..1000)

process echo_sleep {
  input:
  val x from num

	output:
	stdout result

  "echo $x && sleep 1"
}

result.subscribe { println it }

workflow.onComplete {
    println "Pipeline completed at: $workflow.complete"
    println "Execution status: ${ workflow.success ? 'OK' : 'failed' }"
}

nextflow.config:

process {
  executor='lsf'
  queue='normal'
	memory='100MB'
}

install wr:

wget https://github.com/VertebrateResequencing/wr/releases/download/v0.17.0/wr-linux-x86-64.zip
unzip wr-linux-x86-64.zip
mv wr /to/somewhere/in/my/PATH/wr

run:

wr manager start -s lsf
echo "nextflow run ./echo_1000_sleep.nf" | wr add --bsub -r 0 -i nextflow --cwd_matters --memory 1GB

Here's what I did to get it to work in OpenStack...

nextflow_install.sh:

sudo apt-get update
sudo apt-get install openjdk-8-jre-headless -y
wget -qO- https://get.nextflow.io | bash
sudo mv nextflow /usr/bin/nextflow

put input files in S3:

s3cmd put nextflow.config s3://sb10/nextflow/nextflow.config
s3cmd put echo_1000_sleep.nf s3://sb10/nextflow/echo_1000_sleep.nf

~/.openstack_rc:

[your rc file containing OpenStack environment variables downloaded from Horizon]

run:

source ~/.openstack_rc
wr cloud deploy --os 'Ubuntu Xenial' --username ubuntu
echo "cp echo_1000_sleep.nf /shared/echo_1000_sleep.nf && cp nextflow.config /shared/nextflow.config && cd /shared && nextflow run echo_1000_sleep.nf" | wr add --bsub -r 0 -o 2 -i nextflow --memory 1GB --mounts 'ur:sb10/nextflow' --cloud_script nextflow_install.sh --cloud_shared

The NFS share at /shared created by the --cloud_shared option is slow and limited in size; a better solution would be to set up your own high performance shared filesystem in OpenStack (eg. GlusterFS), then add to nextflow_install.sh to mount this share. Or even better, is there a way to have Nextflow not store state on disk? If it could just query wr for job completion status, that would be better.

[pditommaso]

This is a nice initiative, I'll be very happy to review and merge a pull request implementing the support for this back-end engine.

[sb10]

@pditommaso Is there a way to get around needing a shared disk for Nextflow state?
If I was going to try and write this myself, despite not knowing Java, what's my best starting point? Base it on Nextflow's LSF scheduler? Or one of the cloud ones? Has anyone written a guide to writing Nextflow schedulers?

[pditommaso]

Is there a way to get around needing a shared disk for Nextflow state

Not easily, the bottom line is that temporary task outputs need to be shared to downstream tasks, therefore some kind of shared storage. For example with AWS Batch NF is using S3 as shared storage, I guess for OpenStack there's a similar object storage solution (and actually there's).

what's my best starting point?

It depends, if wr provides a command line tool somehow similar to qsub, bsub, etc, your can just create a executor extending AbstractGridExecutor along the same lines of the one for Slurm for example.

If you are planning instead to use a low-level API, you can have a look can extend the Executor base case.

Has anyone written a guide to write Nextflow schedulers

Nope, but I can advise you if needed.

[sb10]

temporary task outputs need to be shared to downstream tasks, therefore some kind of shared storage. For example with AWS Batch NF is using S3 as shared storage

That works for wr's purposes; wr provides easy mounting of S3, which could be used for job inputs/outputs.

The problem I have is with Nextflow running on the "master" node needing to check the output written to S3 on a "worker" node before it, eg. submits the next job.

wr's S3 mount is not POSIX enough for what Nexflow on the "master" needs to do for this state checking. (eg. it caches S3 "directory" contents, so is not aware if a different process writes to that "directory")

Can job input/outputs be stored in S3, but job state be queried from the scheduler?

[pditommaso]

Can job input/outputs be stored in S3, but job state be queried from the scheduler?

Yes, you will need to write your own version of TaskHandler

[sb10]

I have an initial basic but working implementation here: https://github.com/sb10/nextflow/blob/wr/modules/nextflow/src/main/groovy/nextflow/executor/WrExecutor.groovy

I have # *** comments for areas where I'm stuck. Can you give me some pointers? In particular:

1. Don't know how to pick up options from nextflow.config.
2. Not sure as to the best approach for doing the important things BashWrapperBuilder does without creating a file.
3. How do you get a "pure" task name?
4. How do you get Groovy to trust a ca cert?
5. Not sure how passing all possible task options should be done. Just pass around the task object, and will that have all desirable information like memory requirement hints etc?
6. Under what circumstances does kill() get called?

I've only had a quick glance at an example LSF test, and didn't really get it. How should I go about writing a complete test for this? Are end-to-end tests done, or only mocks?

I'm currently using the TaskPollingMonitor, but the ideal way of implementing this would be for Nextflow to listen on a websocket and for wr to tell it the moment that tasks change state. What would be the best approach to doing that? Implement my own TaskMonitor? Or is it even possible?

Currently, my implementation is only working running locally. I still haven't thought about how to handle input and output files being in S3. With existing Executors, how does a user start a workflow running when their input files are not available locally, but in an S3 bucket? And how do they specify that the output files go to S3 as well?

[pditommaso]

It looks a good start.

I have # *** comments for areas where I'm stuck.

I would suggest submitting a draft PR so I can reply to your comments.

Don't know how to pick up options from nextflow.config.

The Executor object has a session attribute that holds the config map. Something like session.config.navigate('your.attribute') should work.

Not sure as to the best approach for doing the important things BashWrapperBuilder does without creating a file.

What do you mean?

How do you get a "pure" task name?

What do you mean for pure?

How do you get Groovy to trust a ca cert?

More than Groovy, you should look at URLConnection Java API. Have a look at K8sClient

Not sure how passing all possible task options should be done. Just pass around the task object, and will that have all desirable information like memory requirement hints etc?

Task specific settings are accessible on TaskRun#config map. Have a look for example here.

Under what circumstances does kill() get called?

Mostly when the execution is aborted and the cleanup method is invoked.

How should I go about writing a complete test for this? Are end-to-end tests done, or only mocks?

Mocks would be a good start. Have a look a K8sClientTest

I'm currently using the TaskPollingMonitor, but the ideal way of implementing this would be for Nextflow to listen on a websocket and for wr to tell it the moment that tasks change state.

I would keep this as a future enhancement.

Currently, my implementation is only working running locally. I still haven't thought about how to handle input and output files being in S3. With existing Executors, how does a user start a workflow running when their input files are not available locally, but in an S3 bucket?

This should be managed automatically by BashWrapperBuilder

[sb10]

I have # *** comments for areas where I'm stuck.

I would suggest submitting a draft PR so I can reply to your comments.

PR created here: #1114

Don't know how to pick up options from nextflow.config.

The Executor object has a session attribute that holds the config map. Something like session.config.navigate('your.attribute') should work.

OK, I'll give it a try. But what is session.getConfigAttribute() and how are you supposed to use it?

Not sure as to the best approach for doing the important things BashWrapperBuilder does without creating a file.

What do you mean?

If possible I don't want Nextflow to create any files of its own (only tasks commands should create workflow output files). If that's not possible, then at least I need the wrapper script it creates to not have any absolute paths in it, because those paths won't exist on other OpenStack instances that wr might create to run tasks.

How do you get a "pure" task name?

What do you mean for pure?

For a scatter task, task.name looks like name (1), while I just want name.

Under what circumstances does kill() get called?

Mostly when the execution is aborted and the cleanup method is invoked.

Specifically, what state could a task be in when kill() is called? Any state?

Currently my code only considers "started" and "complete" statuses, which actually comprise multiple underlying states, and miss some possibilities. For example, in wr a job can be "lost" if it was running on an instance that got deleted externally. The user confirms the situation and then tells wr the job is really dead, and can then choose to restart it.

I imagine the current situation with my Nextflow implementation, it would initially satisfy checkIfRunning(), but never satisfy checkIfCompleted(). Is something going to happen automatically in Nextflow to handle these situations, or should I either a) include "lost" as a "complete" status, or b) should users be checking wr's own status interface to deal with lost jobs?

Currently, my implementation is only working running locally. I still haven't thought about how to handle input and output files being in S3. With existing Executors, how does a user start a workflow running when their input files are not available locally, but in an S3 bucket?

This should be managed automatically by BashWrapperBuilder

It's not clear to me how this is done. What does a user have to do to enable this feature?

The scenario I envisage is:

Nextflow is run on a local machine from which wr has been deployed to OpenStack. This local machine does not have the S3 bucket mounted, and is completely different to the OpenStack servers that will be created to run tasks (in particular, different home directory location).

nextflow.config is configured with the location of an S3 bucket that holds their input files and where output files should go, and their ~/.s3cfg holds their connection details.

This results in WrRestApi.add() (called by submit()) running the TaskBean(task).script in a directory that is mounted on the configured S3 bucket location.

For this to actually work, task.getWorkDirStr() would have to return a unique task path relative to the configured S3 bucket location, and wr would cd there before running the task bean script. Or alternatively, copyStrategy.getStageInputFilesScript needs to symlink from the correct location within a mounted bucket location to the local working directory, and then output files need to be copied over to the mounted bucket location.

Is a scenario like this at all possible? Or what is the "Nextflow" way of doing this kind of thing?

[pditommaso]

OK, I'll give it a try. But what is session.getConfigAttribute() and how are you supposed to use it?

Actually, it should work. TES executor uses the same method. The best thing to do is to create a unit test and debug it.

If possible I don't want Nextflow to create any files of its own (only tasks commands should create workflow output files). If that's not possible, then at least I need the wrapper script it creates to not have any absolute paths in it, because those paths won't exist on other OpenStack instances that wr might create to run tasks.

You can create a copy strategy that does nothing. Have a look at TesFileCopyStrategy

For a scatter task, task.name looks like name (1), while I just want name

Check the TaskRun#processor#name

Specifically, what state could a task be in when kill() is called? Any state?

It's used to kill pending or running tasks when the workflow execution is aborted.

It's not clear to me how this is done. What does a user have to do to enable this feature?

It strongly depends on the storage type used as a pipeline work dir. When an input file is stored in a file system different for the one used as work dir, NF copies that file from the foreign file system to the work dir (for example, an input is on ftp and the work dir is a S3 bucket). This is controlled by the resolveForeignFiles method linked above. Then if the work dir is NOT a posix file system, files are copied to a scratch node in the local node. This is done generally by the Task handler.

[sb10]

Are there any guides or examples aimed at end-users of using Nextflow with data in S3?

[pditommaso]

There's nothing special, file paths need to start with s3://