Integrate SpecificationCluster and adaptive logic updates from Distributed

[guillaumeeb]

dask/distributed#2675 and dask/distributed#2735 introduced some changes and new possibility into how we must handle cluster objects. We need to update dask-jobqueue to be compatible and leverage this update.

See also #280 for a first problem it raises.

[mrocklin]

I would like to experiment with an implementation of dask-jobqueue on top of SpecCluster. I think that we are ready for this now. If possible, I would like help, both to share work, and also to make sure other maintainers feel at ease with this design and want to develop it in the future. With that goal, I propose a plan here and hopefully other people can help to implement pieces of it.

A Worker Job Class

Rather than focus around a Cluster class, lets focus around a class for a single Job. This will allow us both to compose with SpecCluster, and to support more heterogeneity in the future. We need this class to have a specific API, which I outline below:

class Job:
    """
    An interface for Scheduler and Worker processes for use in SpecCluster

    This interface is responsible to submit a worker or scheduler process to a
    resource manager like Kubernetes, Yarn, or SLURM/PBS/SGE/...
    It should implement the methods below, like ``start`` and ``close``
    """
    async def start(self):
        """ Submit the process to the resource manager

        For workers this doesn't have to wait until the process actually starts,
        but can return once the resource manager has the request, and will work
        to make the job exist in the future

        For the scheduler we will expect the scheduler's ``.address`` attribute
        to be avaialble after this completes.
        """
        # this will call `qsub/sbatch`
        self.status = "running"

    async def close(self):
        """ Close the process

        This will be called by the Cluster object when we scale down a node,
        but only after we ask the Scheduler to close the worker gracefully.
        This method should kill the process a bit more forcefully and does not
        need to worry about shutting down gracefully
        """
        # this will call `qdel/...`
        self.status = "closed"

(I took these docstrings from the ProcessInterface class in https://github.com/dask/distributed/blob/master/distributed/deploy/spec.py)

So we would build a new generic class with a constructor that handled all of the keyword arguments, and generated a job script. We would probably just copy and paste the current code from the core JobQueueCluster, but remove out all of the cluster handling.

Specialize this class to particular job queuing systems

We would then need to make specific implementations of this for PBS, SLURM, SGE, and so on. My guess is that this involves copy-pasting from the existing implementations.

At this point something like the following should work:

async def f():
    async with Scheduler() as s:
        async with Client(s.address, asynchronous=True) as client:
            job = PBSJob(project=..., queue=..., ...)
            job = await job
            # verify that this job shows up in qstat (or similar)
            await client.wait_for_workers(1)
            await job.close()

import asyncio
asyncio.get_event_loop().run_until_complete(f())

Build a function to construct a SpecCluster

This should be relatively straightforward. The SSH equivalent is here as a model:

https://github.com/dask/distributed/blob/b68660cb873e8efb9d9b47a92e39ab78f2fd7573/distributed/deploy/ssh2.py#L158-L210

Generalize the Job class for the scheduler

The code in SpecCluster doesn't require the scheduler to run locally. We could adapt/subclass the Job class to also run a Scheduler, and then include that in SpecCluster as an alternative to having a local Scheduler object.

[lesteve]

I am willing to offer a bit of help to see this go forward but that won't be much more than 1-2 hours per week I am afraid.

[mrocklin]

For a current maintainer I estimate getting the initial Job class working smoothly will take a day, and then moving over the other cluster managers will take 1-2 hours each.

…

On Mon, Aug 5, 2019 at 5:36 AM Loïc Estève ***@***.***> wrote: I am willing to offer a bit of help to see this go forward but that won't be much more than 1-2 hours per week I am afraid. — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#281?email_source=notifications&email_token=AACKZTB62ISE7N367EENPYLQDANEVA5CNFSM4H3QKTC2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD3RVVIQ#issuecomment-518216354>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AACKZTCLYHR4Q2LFTFXWBBDQDANEVANCNFSM4H3QKTCQ> .

[mrocklin]

An early attempt at the first step of creating a Job class is here: #307

I haven't really tested it yet though. I'm writing this in a little bit of a rush.

[lesteve]

Thanks a lot for starting this, I'll try to have a look shortly!

[guillaumeeb]

Thanks a lot Matt! I'm in too, but in holyday for the time being. Probably not of much help until end of august...

[guillaumeeb]

Fixed by #307. Thanks again @mrocklin and @lesteve