K8s operator for RayServe

[ashahab]

Hi, could you help me create a K8S operator using kubebuilder? Here are what it can do: This K8S operator listens for a specific event. Once it receives an event, it will download a protobuf file from a specific API using bearer authentication. It will then parse the protobuf file's contents into a Ray DAG. It will then create a RayServe job config from the Ray DAG. Then it will create a RayServe job using the RayServe job config for the KubeRay operator.

[alabulei1]

Sure, I'd be happy to help you create a K8S operator for RayServe using Kubebuilder! Here's an outline of the steps you can follow:

1. Install Kubebuilder and set your development environment.
2. Create a new Kubebuilder project using the kubebuilder init command.
3. Create a new API by defining a new kind in the api/v1 directory.
4. Define the event you want the operator to listen to in the api/v1/event_types.go file.
5. Define the controller logic in the controllers directory. This is where you will download the protobuf file, parse its contents into a Ray DAG, create a RayServe job config, and create a RayServe job using the KubeRay operator API.
6. Add custom resource definition (CRD) in config/samples directory.
7. Generate code and add the controller to the manager.
8. Build and deploy the operator.

It may be helpful to check the Kubebuilder documentation for more information on how to create a K8S operator. Good luck!

[ashahab]

Can you elaborate on the following: "Define the controller logic in the controllers directory. This is where you will download the protobuf file, parse its contents into a Ray DAG, create a RayServe job config, and create a RayServe job using the KubeRay operator API."

[alabulei1]

Sure, here is a more detailed explanation of the steps you can take to define the controller logic:

1. Download the Protobuf file: Use the Kubernetes API to listen for the specific event that you defined in event_types.go. Once it triggers, you can download the Protobuf file using a bearer token authentication from the specific API service.

2. Parse the contents into a Ray DAG: After downloading the Protobuf file, you need to parse its contents into a Ray DAG. In order to do that, you need to convert the Protobuf message into a Ray DAG object. You can use the protocol buffers library to parse the message and create the Ray DAG object.

3. Create a RayServe job config: After parsing the Ray DAG, you need to create a RayServe job configuration that includes the configuration settings required for the RayServe job. This includes defining the RayServe serving function, the resources needed by the RayServe job, and other relevant configuration options.

4. Create a RayServe job: After creating the job configuration, you can create a RayServe job using the KubeRay operator API. You will need to send a POST request to the KubeRay operator API endpoint, and include the job configuration in the request body.

In order to define the controller logic for your K8S operator, you will need to create a new Golang file in the controllers directory. This file will contain the main logic for handling the event and performing the above steps. You can define the various functions required for each step in this file, and then call those functions when the event is triggered. Within the Kubebuilder framework, you can add various hooks that get triggered when a new event occurs, and you can add custom logic to these hooks to carry out the specific operations you require.

I hope that helps! Let me know if you have any further questions.

[ashahab]

What would the reconciliation loop of the controller do?

[alabulei1]

The reconciliation loop of the controller is responsible for ensuring that the desired state of the system matches the actual state of the system. It does this by continuously monitoring the Kubernetes resources and the state of the RayServe jobs, and taking corrective action if necessary to ensure that the system is in the correct state.

For the specific K8S operator you're building to create and manage RayServe jobs, the reconciliation loop would do the following:

1. Check if there is a new event triggering the creation/modification/deletion of a custom resource for your operator. If there is no new event, the controller will wait for a new event to occur before continuing.

2. Once a new event is detected, the controller will download the protobuf file from the specified API using bearer authentication.

3. Next, the controller will parse the protobuf into a Ray DAG and create a RayServe job configuration using the parsed Ray DAG.

4. The controller will then check the actual state of the system to see if the RayServe job already exists. If the RayServe job already exists with the same name, the controller will check if there are any differences between the existing RayServe job and the new RayServe job configuration. If there are any differences that need to be made to the existing RayServe job, the controller will update the existing RayServe job with the new configuration.

5. If the RayServe job does not already exist, the controller will create a new RayServe job using the KubeRay operator API.

6. Finally, the controller will update the status of the custom resource for the RayServe job to reflect the current state of the system.

The reconciliation loop will continue to monitor the state of the system, and will take corrective action if necessary to ensure that the desired state of the system is maintained. If there are any changes to the custom resource, it will trigger a new event, and the reconciliation loop will start again from step 1.

[ashahab]

Can you implement the reconciliation loop in golang?

[alabulei1]

Sure, here is an example implementation of the reconciliation loop for a K8S operator that creates and manages RayServe jobs using the Kubebuilder framework:

// Reconcile function is invoked by the Kubernetes API to achieve the desired state of the system.
func (r *RayServeReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
    // Obtain context and logger.
    ctx := context.Background()
    log := r.Log.WithValues("rayserve", req.NamespacedName)

    // Fetch the RayServe object for the given request.
    rs := &operatorv1alpha1.RayServe{}
    err := r.Get(ctx, req.NamespacedName, rs)
    if err != nil {
        // If the RayServe object is not found, ignore it.
        if k8serrors.IsNotFound(err) {
            log.Info("RayServe resource not found, ignoring")
            return ctrl.Result{}, nil
        }
        // Return error when there is something wrong with fetching the RayServe.
        log.Error(err, "failed to fetch RayServe resource")
        return ctrl.Result{}, err
    }

    // Create a client to access the KubeRay operator API.
    client, err := kuberay.NewClient(ctx, r.Client, r.Scheme, rs.Namespace)
    if err != nil {
        log.Error(err, "failed to create KubeRay client")
        return ctrl.Result{}, err
    }

    // Download the protobuf file using bearer authentication from the specified API.
    pbFile, err := downloadProtobufFile(ctx, rs)
    if err != nil {
        log.Error(err, "failed to download protobuf file")
        return ctrl.Result{}, err
    }

    // Parse the protobuf file into a Ray DAG.
    rayDag, err := parseProtobufIntoRayDag(pbFile)
    if err != nil {
        log.Error(err, "failed to parse protobuf into Ray DAG")
        return ctrl.Result{}, err
    }

    // Create the RayServe job configuration using the parsed Ray DAG.
    rsJob, err := createRayServeJobFromRayDag(rs.Name, rs.Namespace, rayDag)
    if err != nil {
        log.Error(err, "failed to create RayServe job from Ray DAG")
        return ctrl.Result{}, err
    }

    // Check if the RayServe job already exists.
    existingJob, err := client.GetJob(ctx, rs.Name)
    if err != nil && !kuberay.IsNotFound(err) {
        log.Error(err, "failed to check if RayServe job exists")
        return ctrl.Result{}, err
    }

    // If the RayServe job already exists, check if there are updates to be made.
    if existingJob != nil {
        if !reflect.DeepEqual(rsJob, existingJob.Config) {
            log.Info("updating RayServe job configuration")
            if err := client.DeleteJob(ctx, rs.Name); err != nil {
                log.Error(err, "failed to delete existing RayServe job")
                return ctrl.Result{}, err
            }
            if err := client.CreateJob(ctx, rsJob); err != nil {
                log.Error(err, "failed to create new RayServe job after deleting existing")
                return ctrl.Result{}, err
            }
        }
    } else {
        // Otherwise, create a new RayServe job.
        log.Info("creating RayServe job configuration")
        if err := client.CreateJob(ctx, rsJob); err != nil {
            log.Error(err, "failed to create RayServe job")
            return ctrl.Result{}, err
        }
    }

    // Update the status of the RayServe resource to reflect the current state of the RayServe job.
    rs.Status.JobConfig = rsJob
    if err := r.Status().Update(ctx, rs); err != nil {
        log.Error(err, "failed to update RayServe status")
        return ctrl.Result{}, err
    }

    return ctrl.Result{}, nil
}

This implementation assumes that you have implemented functions for downloading the protobuf file, parsing the protobuf into a Ray DAG, and creating a RayServe job configuration from the Ray DAG.

The RayServeReconciler is the main controller for the RayServe resource, and the Reconcile function is the main reconciliation loop for this controller. The reconciliation loop checks the current state of the RayServe resource, downloads the protobuf file, parses it into a Ray DAG, creates a RayServe job configuration from the Ray DAG, and then creates or updates the RayServe job using the KubeRay client. Finally, it updates the status of the RayServe resource to reflect the current state of the RayServe job. The reconciliation loop will continue to run periodically to ensure that the desired state of the system is maintained.