Ray operator: controller code and guide to use (#6501)

deploy/ray-operator/controllers/common/constant.go

@@ -0,0 +1,33 @@
+package common
+
+const (
+	// Head used as pod type to decide create service or not, for now only create service for head.
+	Head = "head"
+
+	// Belows used as label key
+	//rayclusterComponent is the pod name for this pod for selecting pod by pod name.
+	rayclusterComponent = "raycluster.component"
+	// rayIoComponent is the identifier for created by ray-operator for selecting pod by operator name.
+	rayIoComponent = "rayclusters.ray.io/component-name"
+	// RayClusterOwnerKey is the ray cluster instance name for selecting pod by instance name.
+	RayClusterOwnerKey = "raycluster.instance.name"
+	// ClusterPodType is the pod type label key for selecting pod by type.
+	ClusterPodType = "raycluster.pod.type"
+
+	// rayOperator is the value of ray-operator used as identifier for the pod
+	rayOperator = "ray-operator"
+
+	// Use as separator for pod name, for example, raycluster-small-size-worker-0
+	DashSymbol = "-"
+
+	// Use as default port
+	defaultHTTPServerPort = 30021
+	defaultRedisPort      = 6379
+
+	// Check node if ready by checking the path exists or not
+	PodReadyFilepath = "POD_READY_FILEPATH"
+
+	// Use as container env variable
+	namespace   = "NAMESPACE"
+	clusterName = "CLUSTER_NAME"
+)

deploy/ray-operator/controllers/common/meta.go

@@ -0,0 +1,18 @@
+package common
+
+import rayiov1alpha1 "ray-operator/api/v1alpha1"
+
+// The function labelsForCluster returns the labels for selecting the resources
+// belonging to the given RayCluster CR name.
+func labelsForCluster(instance rayiov1alpha1.RayCluster, name string, podTypeName string, extend map[string]string) (ret map[string]string) {
+	ret = map[string]string{
+		rayclusterComponent: name,
+		rayIoComponent:      rayOperator,
+		RayClusterOwnerKey:  instance.Name,
+		ClusterPodType:      podTypeName,
+	}
+	for k, v := range extend {
+		ret[k] = v
+	}
+	return
+}

deploy/ray-operator/controllers/common/pod.go

@@ -0,0 +1,129 @@
+package common
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	rayiov1alpha1 "ray-operator/api/v1alpha1"
+	"strings"
+)
+
+type PodConfig struct {
+	RayCluster  *rayiov1alpha1.RayCluster
+	PodTypeName string
+	PodName     string
+	Extension   rayiov1alpha1.Extension
+}
+
+func DefaultPodConfig(instance *rayiov1alpha1.RayCluster, podTypeName string, podName string) *PodConfig {
+	return &PodConfig{
+		RayCluster:  instance,
+		PodTypeName: podTypeName,
+		PodName:     podName,
+	}
+}
+
+// Build a pod for the cluster instance.
+func BuildPod(conf *PodConfig) *corev1.Pod {
+	// build label for cluster
+	rayLabels := labelsForCluster(*conf.RayCluster, conf.PodName, conf.PodTypeName, conf.Extension.Labels)
+
+	// build container for pod, now only handle one container for each pod
+	var containers []corev1.Container
+	container := buildContainer(conf)
+	containers = append(containers, container)
+
+	// create volume
+	volumes := conf.Extension.Volumes
+
+	spec := corev1.PodSpec{
+		Volumes:            volumes,
+		Containers:         containers,
+		Affinity:           conf.Extension.Affinity,
+		Tolerations:        conf.Extension.Tolerations,
+		ServiceAccountName: conf.RayCluster.Namespace,
+	}
+
+	// build annotations and store podCompareHash for comparison
+	annotations := conf.Extension.Annotations
+
+	pod := &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Pod",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        conf.PodName,
+			Namespace:   conf.RayCluster.Namespace,
+			Labels:      rayLabels,
+			Annotations: annotations,
+		},
+		Spec: spec,
+	}
+
+	return pod
+}
+
+// Build container for pod.
+func buildContainer(conf *PodConfig) corev1.Container {
+
+	redisPort := defaultRedisPort
+	httpServerPort := defaultHTTPServerPort
+	jobManagerPort := defaultRedisPort
+
+	// get pod file path to check if the pod container ready or not
+	var podReadyFilepath string
+	for _, env := range conf.Extension.ContainerEnv {
+		if strings.EqualFold(env.Name, PodReadyFilepath) {
+			podReadyFilepath = env.Value
+			break
+		}
+	}
+
+	// assign image by typeName
+	image := conf.RayCluster.Spec.Images.DefaultImage
+	if conf.Extension.Image != "" {
+		image = conf.Extension.Image
+	}
+
+	volumeMounts := conf.Extension.VolumeMounts
+
+	// add instance name and namespace to container env to identify cluster pods
+	var containerEnv []corev1.EnvVar
+	containerEnv = conf.Extension.ContainerEnv
+	containerEnv = append(containerEnv,
+		corev1.EnvVar{Name: namespace, Value: conf.RayCluster.Namespace},
+		corev1.EnvVar{Name: clusterName, Value: conf.RayCluster.Name})
+
+	container := corev1.Container{
+		Name:            strings.ToLower(conf.PodTypeName),
+		Image:           image,
+		Command:         []string{"/bin/bash", "-c", "--"},
+		Args:            []string{conf.Extension.Command},
+		Env:             containerEnv,
+		Resources:       conf.Extension.Resources,
+		VolumeMounts:    volumeMounts,
+		ImagePullPolicy: conf.RayCluster.Spec.ImagePullPolicy,
+		Ports: []corev1.ContainerPort{
+			{
+				ContainerPort: int32(redisPort),
+				Name:          "redis",
+			},
+			{
+				ContainerPort: int32(httpServerPort),
+				Name:          "http-server",
+			},
+			{
+				ContainerPort: int32(jobManagerPort),
+				Name:          "job-manager",
+			},
+		},
+		ReadinessProbe: &corev1.Probe{
+			Handler: corev1.Handler{
+				Exec: &corev1.ExecAction{Command: []string{"cat", podReadyFilepath}},
+			},
+			InitialDelaySeconds: 15,
+			SuccessThreshold:    2,
+		},
+	}
+	return container
+}

deploy/ray-operator/controllers/common/service.go

@@ -0,0 +1,45 @@
+package common
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	rayiov1alpha1 "ray-operator/api/v1alpha1"
+	"ray-operator/controllers/utils"
+	"strings"
+)
+
+type ServiceConfig struct {
+	RayCluster rayiov1alpha1.RayCluster
+	PodName    string
+}
+
+func DefaultServiceConfig(instance rayiov1alpha1.RayCluster, podName string) *ServiceConfig {
+	return &ServiceConfig{
+		RayCluster: instance,
+		PodName:    podName,
+	}
+}
+
+// Build service for pod, for now only head pod will have service.
+func ServiceForPod(conf *ServiceConfig) *corev1.Service {
+	name := conf.PodName
+	if strings.Contains(conf.PodName, Head) {
+		name = utils.Before(conf.PodName, Head) + "head"
+	}
+
+	svc := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: conf.RayCluster.Namespace,
+		},
+		Spec: corev1.ServiceSpec{
+			ClusterIP: "None",
+			// select this raycluster's component
+			Selector: map[string]string{
+				rayclusterComponent: conf.PodName,
+			},
+		},
+	}
+
+	return svc
+}