Environ plugin: also update dask config

[gjoseph92]

Because Nannies pass their config into worker through a separate config= argument, setting dask config environment variables with the Enviorn nanny plugin confusingly had no effect. Now, any dask-config environment variables work automatically (besides memory limit, which is still handled specially by Nanny).

• Closes #xxxx
• Tests added / passed
• Passes black distributed / flake8 distributed / isort distributed

[jrbourbeau]

Thanks @gjoseph92!

It looks like this behavior can happen with the Nanny, independent of the Environ plugin:

distributed/distributed/nanny.py

Lines 820 to 821 in 332f785

	os.environ.update(env)
	dask.config.set(config)

Updating in WorkerProcess._run seems like a more centralized place to put this type of logic.

Also, can you add a test which demonstrates things are working as expected? See, for example, distributed/tests/test_nanny.py::test_environment_variable_by_config

[gjoseph92]

Sure, that makes sense to me. The interface for specifying config/environ to the Nanny seems a bit ambiguous, since there are a number of overlapping options (distributed.nanny.environ vs env= vs config= vs anything already set in dask.config.config via the actual process environment/YAMLs). But if having env= (but not anything in the actual os.environ) potentially override config= is desirable, that is a more general solution.

There were no tests for the Environ plugin to begin with so I continued the trend :) I'll add them for the changes to Nanny itself though.

[jrbourbeau]

Yeah, prioritization here can get tricky. I think there was a previous conversation around this, but I can't seem to find it. cc @jacobtomlinson who may have thoughts on this topic

There's this existing tests which could serve as a good model

distributed/distributed/tests/test_nanny.py

Lines 601 to 611 in 2298861

	@gen_cluster(client=True, Worker=Nanny, timeout=10000000)
	async def test_environ_plugin(c, s, a, b):
	from dask.distributed import Environ
	await c.register_worker_plugin(Environ({"ABC": 123}))
	async with Nanny(s.address, name="new") as n:
	results = await c.run(os.getenv, "ABC")
	assert results[a.worker_address] == "123"
	assert results[b.worker_address] == "123"
	assert results[n.worker_address] == "123"

[gjoseph92]

There's this existing tests which could serve as a good model

🤦 because the plugin is in plugin.py, I just looked in test_plugin.py, not test_nanny.py