Ray cluster CRD and example CR + multi-ray-cluster operator

[DmitriGekhtman]

Why are these changes needed?

Following up on #11929, this PR adds a k8s CRD describing a ray cluster configuration and an example ray cluster CR. A ray cluster CR is pretty much just a reformatted version of one of the current ray cluster configs.

This PR also extends the operator such that it can manage multiple Ray clusters.
Using kubectl, users can create/update/delete clusters and check monitoring logs.

---

Related issue number

#11929 #11545

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 :(

Create, update, delete, logging working as expected for me locally.

[DmitriGekhtman]

Currently debugging the following error, which occurs after starting a cluster, shutting it down, and trying to start a new cluster by applying, deleting, applying a raycluster custom resource. The error takes place immediately after the monitor initializes a StandardAutoscaler.

edit: Takes place in Monitor.update_raylet_map().

edit: This probably happens because we can only support one GlobalState per interpreter session.

edit: Running Monitors in subprocesses instead of threads could solve this -- going to try that.

[2020-11-30 08:22:38,240 C 27 87] service_based_gcs_client.cc:207: Couldn't reconnect to GCS server. The last attempted GCS server address was :0
*** StackTrace Information ***
    @     0x7f74211423e5  google::GetStackTraceToString()
    @     0x7f74210b704e  ray::GetCallTrace()
    @     0x7f74210dc454  ray::RayLog::~RayLog()
    @     0x7f7420d2952a  ray::gcs::ServiceBasedGcsClient::ReconnectGcsServer()
    @     0x7f7420d295ad  ray::gcs::ServiceBasedGcsClient::GcsServiceFailureDetected()
    @     0x7f7420d3228f  _ZNSt17_Function_handlerIFvRKN3ray6StatusERKNS0_3rpc19GetAllNodeInfoReplyEEZNS4_12GcsRpcClient14GetAllNodeInfoERKNS4_21GetAllNodeInfoRequestERKSt8functionIS8_EEUlS3_S7_E_E9_M_invokeERKSt9_Any_dataS3_S7_
    @     0x7f7420d37115  ray::rpc::ClientCallImpl<>::OnReplyReceived()
    @     0x7f7420c4cb12  _ZN5boost4asio6detail18completion_handlerIZN3ray3rpc17ClientCallManager29PollEventsFromCompletionQueueEiEUlvE_E11do_completeEPvPNS1_19scheduler_operationERKNS_6system10error_codeEm
    @     0x7f74211afa51  boost::asio::detail::scheduler::do_run_one()
    @     0x7f74211b06a9  boost::asio::detail::scheduler::run()
    @     0x7f74211b2a07  boost::asio::io_context::run()
    @     0x7f7420c15e44  _ZNSt6thread11_State_implINS_8_InvokerISt5tupleIJZN3ray3gcs19GlobalStateAccessorC4ERKSsS7_bEUlvE_EEEEE6_M_runEv
    @     0x7f7421455120  execute_native_thread_routine
    @     0x7f7422194609  start_thread
    @     0x7f74220bb293  clone

[DmitriGekhtman]

Running the monitor in a subprocess fixed the previous issue -- the code is now functional!
To-do list in PR description.

[yiranwang52]

is it possible for the subprocesses to be leaked if operator.py is killed unexpectedly?

[DmitriGekhtman]

I think currently if operator.py is killed unexpectedly, the operator pod will shut down.
Which reminds me -- I was going to make all of the ray-clusters managed by the operator fate-share with the operator.
So then everything would go down.
That's of course not optimal behavior.

Let me also experiment to see what happens if operator.py is killed unexpectedly but the pod doesn't go down.

[yiranwang52]

I don't think you want to kill all the clusters when operator go down.
It just need to know about them when started.

[DmitriGekhtman]

Ah, right. I think that should work with the code as it is. (When the operator restarts it should create_or_update on each cluster.)
Will test.

[DmitriGekhtman]

I think Python does an OK job of cleaning up processes spawned by the multiprocessing module?
To check, I ran a test script that uses multiprocessing to spawn a dummy process that runs forever. After doing a Ctrl-C keyboard interrupt, the pid of the process is no longer present in the output of ps -ef.

I'll check that the monitor processes behave in the same way.

Let me know if there's something that can be done to ensure that the processes are cleaned correctly.
(besides rewriting everything to have each cluster's autoscaler .update in a for loop, which is probably a good idea to implement sooner or later or sooner to replace this subprocess logic)

[yiranwang52]

Your test sounds good enough.

[DmitriGekhtman]

Actually, you were right -- terminating operator.py did leak the monitor process. (In my test script, the child process was receiving the keyboard interrupt.)

I've now set the monitor subprocess to be a daemon, and that works -- when I run kill -SIGTERM <operator_pid> it stops the monitor subprocess too.
(If you do kill -SIGKILL <operator_pid> then the daemon monitor subprocess still leaks.)

[yiranwang52]

okay then this is still a problem we should fix in the near future.