Scheduller add_plugin doesn't call start on the plug in

[davzucky]

What happened:
When calling on the Scheduller the method add_plugin the method start on the SchedullerPlugin is not called.

What you expected to happen:
I expect the method start to be called. This is what is happening to the plugins set from the __init___ method of the scheduler. However I'm not sure this is expected behavior or bug. Please confirm why they are this discrepancy, as I would expect the scheduler to call the start method on all the plugins.

Minimal Complete Verifiable Example:

@pytest.mark.asyncio
async def test_k8s_plugin_initialize_cluster(k8s_cluster, docker_image):
    scheduler = Scheduler()
    await scheduler.start()

    pod_spec = make_pod_spec(
            docker_image,
            extra_container_config={
                "imagePullPolicy": "IfNotPresent",
                "securityContext": {"runAsUser": 0},
            },
        )
    plugin = KubernetesAutoScaler(pod_spec, 0, 4)
    scheduler.add_plugin(plugin)
    # await plugin.start(scheduler=scheduler)

    assert plugin._cluster is not None
    assert plugin._cluster.status ==Status.running

Anything else we need to know?:
This is for a new plugin I'm writing for kubernetes dask/dask-kubernetes#377

Environment:

• Dask version: Latest
• Python version: All
• Operating System: Na
• Install method (conda, pip, source): Implementing new test

[jrbourbeau]

Thanks for raising an issue @davzucky. I think you're after the Scheduler.register_scheduler_plugin method which will both ensure that the plugin is added to the scheduler and that the plugin's start method is called

[davzucky]

@jrbourbeau Thank you for your reply. I found the method register_scheduler_plugin, however look like this only work for registering plugin from client which are streamed as pickle.
If I pass a plugin instance to register_scheduler_plugin I get

x = <test_plugin.KubernetesAutoScaler object at 0x7f11927feef0>

    def loads(x, *, buffers=()):
        try:
            if buffers:
                return pickle.loads(x, buffers=buffers)
            else:
>               return pickle.loads(x)
E               TypeError: a bytes-like object is required, not 'KubernetesAutoScaler'

If register_scheduler_plugin the right method to use, should it be extended to support instance as well ?

[jrbourbeau]

Ah, thanks for pointing that out. If you have a client available, then I'd recommend Client.register_scheduler_plugin. Otherwise, then Scheduler.add_plugin + separately running the plugin's start method sounds like a reasonable approach. Alternatively, since your directly creating a scheduler object via the Python API you could also create a plugin instance and pass it directly to Scheduler.__init__ with the plugins= keyword

[davzucky]

Yes, As I'm using this code for testing I moved already initialize the plugin to the scheduler using Scheduler.__init__. I just want to highlight the discrepancy on the API layer.

Do you think the method register_scheduler_plugin should be extended to support as well instance of SchedulerPlugin as input.
If you are happy with that I will prepare a PR.