Implement basic runtime env plugin mechanism

[simon-mo]

This PR implements a very basic and experimental version of pluggable runtime environment.

In particular, it supports loading an out-of-tree runtime env plugin that can:

• modify command prefix (i.e. run arbitrary commands before worker commend)
• modify python executable prefix (i.e. prepend to, or modify the python executable for worker)
• modify environment variables (i.e. inject arbitrary environment variables)

There are a lot of limitation at the moment:

• no support for caching environment with uris mechanism.
• no support for garbage collection environment files.

See here for the interface for the plugin you need to write.

Here is a simple example to configure process niceness score using the nice command, falling into the "modify python executable prefix" category.

We will define a plugin inside my_module.py. I would recommend to put this in an actual python package when running in production.
my_module.py:

from ray._private.runtime_env.context import RuntimeEnvContext
from ray._private.runtime_env.plugin import RuntimeEnvPlugin

class MyPlugin(RuntimeEnvPlugin):
    @staticmethod
    def validate(runtime_env_dict: dict): pass

    @staticmethod
    def modify_context(uri: str, runtime_env_dict: dict, ctx: RuntimeEnvContext) -> None:
        plugin_config_dict = runtime_env_dict["plugins"]["my_module.MyPlugin"]
        ctx.py_executable = (
            plugin_config_dict["prefix_command"] + " " + ctx.py_executable)

app.py:

import ray
ray.init(runtime_env=dict(working_dir=".")) # needed if my_module.py is not part of package

@ray.remote
def f():
    import psutil
    return {"nice": psutil.Process().nice()}

   
output = ray.get(
    f.options(
        runtime_env={
            "plugins": {
                "my_module.MyPlugin": { # my Plugin will be dynamically imported!
                    # See https://en.wikipedia.org/wiki/Nice_(Unix)
                    "prefix_command": "nice -n 19",
                }
            }
        }).remote())

print(output)

> python app.py
2021-10-01 16:49:56,637	INFO services.py:1265 -- View the Ray dashboard at http://127.0.0.1:8265
2021-10-01 16:49:56,881	INFO working_dir.py:414 -- gcs://_ray_pkg_d674a1dc7fa916b6689f85a6f5ac9e1a.zip doesn't exist. Create new package with . and None
2021-10-01 16:49:56,883	INFO working_dir.py:425 -- gcs://_ray_pkg_d674a1dc7fa916b6689f85a6f5ac9e1a.zip has been pushed with 2303 bytes
{'nice': 19}

Checks

• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[edoakes]

@architkulkarni FYI

[edoakes]

Looks great! Thanks for getting this in so quickly.

[edoakes]

does this need to be a field of self? prefer to just make it a global constant

[simon-mo]

Merging as windows tests skipped and relevant tests all passes

[pcmoritz]

@krfricke It's possible that with this PR we can do the stdin/stdout capture for the Ray client without modifying core Ray

[edoakes]

@pcmoritz @krfricke I have no context but that sounds cool -- let me know if you need help/have questions!