[Bug] Service (Serve) changing port from 8000 to 9000 doesn't work (#…

…1081)

Service (Serve) changing port from 8000 to 9000 doesn't work

helm-chart/kuberay-operator/crds/ray.io_rayservices.yaml

@@ -11784,6 +11784,8 @@ spec:
                     type: array
                   importPath:
                     type: string
+                  port:
+                    type: integer
                   runtimeEnv:
                     type: string
                 required:

ray-operator/apis/ray/v1alpha1/rayservice_types.go

@@ -58,6 +58,7 @@ type ServeDeploymentGraphSpec struct {
 	ImportPath       string            `json:"importPath"`
 	RuntimeEnv       string            `json:"runtimeEnv,omitempty"`
 	ServeConfigSpecs []ServeConfigSpec `json:"deployments,omitempty"`
+	Port             int               `json:"port,omitempty"`
 }
 
 // ServeConfigSpec defines the desired state of RayService

ray-operator/config/crd/bases/ray.io_rayservices.yaml

@@ -11784,6 +11784,8 @@ spec:
                     type: array
                   importPath:
                     type: string
+                  port:
+                    type: integer
                   runtimeEnv:
                     type: string
                 required:

ray-operator/config/samples/ray-service.different-port.yaml

@@ -0,0 +1,99 @@
+# ray-service.different-port.yaml: By default, RayService exposes port 8000 to serve requests. This sample YAML
+# file demonstrates how to change the port to 9000. To achieve this, follow these steps:
+# (1) Modify `spec.serveConfig.port` to 9000.
+# (2) Modify the container `ray-head`'s `serve` port to 9000.
+apiVersion: ray.io/v1alpha1
+kind: RayService
+metadata:
+  name: rayservice-sample
+spec:
+  serviceUnhealthySecondThreshold: 300 # Config for the health check threshold for service. Default value is 60.
+  deploymentUnhealthySecondThreshold: 300 # Config for the health check threshold for deployments. Default value is 60.
+  serveConfig:
+    importPath: fruit.deployment_graph
+    runtimeEnv: |
+      working_dir: "https://github.com/ray-project/test_dag/archive/41d09119cbdf8450599f993f51318e9e27c59098.zip"
+    deployments:
+      - name: MangoStand
+        numReplicas: 1
+        userConfig: |
+          price: 3
+        rayActorOptions:
+          numCpus: 0.1
+      - name: OrangeStand
+        numReplicas: 1
+        userConfig: |
+          price: 2
+        rayActorOptions:
+          numCpus: 0.1
+      - name: PearStand
+        numReplicas: 1
+        userConfig: |
+          price: 1
+        rayActorOptions:
+          numCpus: 0.1
+      - name: FruitMarket
+        numReplicas: 1
+        rayActorOptions:
+          numCpus: 0.1
+      - name: DAGDriver
+        numReplicas: 1
+        routePrefix: "/"
+        rayActorOptions:
+          numCpus: 0.1
+    port: 9000
+  rayClusterConfig:
+    rayVersion: '2.4.0' # should match the Ray version in the image of the containers
+    ######################headGroupSpecs#################################
+    # Ray head pod template.
+    headGroupSpec:
+      # the following params are used to complete the ray start: ray start --head --block --redis-port=6379 ...
+      rayStartParams:
+        dashboard-host: '0.0.0.0'
+      #pod template
+      template:
+        spec:
+          containers:
+            - name: ray-head
+              image: rayproject/ray:2.4.0
+              resources:
+                limits:
+                  cpu: 2
+                  memory: 2Gi
+                requests:
+                  cpu: 2
+                  memory: 2Gi
+              ports:
+                - containerPort: 6379
+                  name: gcs-server
+                - containerPort: 8265
+                  name: dashboard
+                - containerPort: 10001
+                  name: client
+                - containerPort: 9000
+                  name: serve
+    workerGroupSpecs:
+      # the pod replicas in this group typed worker
+      - replicas: 1
+        minReplicas: 1
+        maxReplicas: 5
+        # logical group name, for this called small-group, also can be functional
+        groupName: small-group
+        rayStartParams: {}
+        # Pod template
+        template:
+          spec:
+            containers:
+              - name: ray-worker # must consist of lower case alphanumeric characters or '-', and must start and end with an alphanumeric character (e.g. 'my-name',  or '123-abc'
+                image: rayproject/ray:2.4.0
+                lifecycle:
+                  preStop:
+                    exec:
+                      command: ["/bin/sh","-c","ray stop"]
+                resources:
+                  limits:
+                    cpu: "1"
+                    memory: "2Gi"
+                  requests:
+                    cpu: "500m"
+                    memory: "2Gi"

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

@@ -108,15 +108,14 @@ func DefaultHeadPodTemplate(instance rayiov1alpha1.RayCluster, headSpec rayiov1a
 	headSpec.RayStartParams = setAgentListPortStartParams(instance, headSpec.RayStartParams)
 
 	initTemplateAnnotations(instance, &podTemplate)
+	rayContainerIndex := getRayContainerIndex(podTemplate.Spec)
 
 	// if in-tree autoscaling is enabled, then autoscaler container should be injected into head pod.
 	if instance.Spec.EnableInTreeAutoscaling != nil && *instance.Spec.EnableInTreeAutoscaling {
 		headSpec.RayStartParams["no-monitor"] = "true"
 		// set custom service account with proper roles bound.
 		// utils.CheckName clips the name to match the behavior of reconcileAutoscalerServiceAccount
 		podTemplate.Spec.ServiceAccountName = utils.CheckName(utils.GetHeadGroupServiceAccountName(&instance))
-
-		rayContainerIndex := getRayContainerIndex(podTemplate.Spec)
 		rayHeadImage := podTemplate.Spec.Containers[rayContainerIndex].Image
 		// Determine the default image to use for the Ray container.
 		autoscalerImage := getAutoscalerImage(rayHeadImage, instance.Spec.RayVersion)
@@ -127,20 +126,14 @@ func DefaultHeadPodTemplate(instance rayiov1alpha1.RayCluster, headSpec rayiov1a
 		podTemplate.Spec.Containers = append(podTemplate.Spec.Containers, autoscalerContainer)
 	}
 
-	isMetricsPortExists := false
-	for _, port := range podTemplate.Spec.Containers[0].Ports {
-		if port.Name == DefaultMetricsName {
-			isMetricsPortExists = true
-			break
-		}
-	}
+	// If the metrics port does not exist in the Ray container, add a default one for Promethues.
+	isMetricsPortExists := utils.FindContainerPort(&podTemplate.Spec.Containers[rayContainerIndex], DefaultMetricsName, -1) != -1
 	if !isMetricsPortExists {
-		// add metrics port for exposing to the promethues stack.
 		metricsPort := v1.ContainerPort{
 			Name:          DefaultMetricsName,
 			ContainerPort: int32(DefaultMetricsPort),
 		}
-		podTemplate.Spec.Containers[0].Ports = append(podTemplate.Spec.Containers[0].Ports, metricsPort)
+		podTemplate.Spec.Containers[rayContainerIndex].Ports = append(podTemplate.Spec.Containers[rayContainerIndex].Ports, metricsPort)
 	}
 
 	return podTemplate
@@ -240,15 +233,9 @@ func DefaultWorkerPodTemplate(instance rayiov1alpha1.RayCluster, workerSpec rayi
 
 	initTemplateAnnotations(instance, &podTemplate)
 
-	isMetricsPortExists := false
-	for _, port := range podTemplate.Spec.Containers[rayContainerIndex].Ports {
-		if port.Name == DefaultMetricsName {
-			isMetricsPortExists = true
-			break
-		}
-	}
+	// If the metrics port does not exist in the Ray container, add a default one for Promethues.
+	isMetricsPortExists := utils.FindContainerPort(&podTemplate.Spec.Containers[rayContainerIndex], DefaultMetricsName, -1) != -1
 	if !isMetricsPortExists {
-		// add metrics port for exposing to the promethues stack.
 		metricsPort := v1.ContainerPort{
 			Name:          DefaultMetricsName,
 			ContainerPort: int32(DefaultMetricsPort),

ray-operator/controllers/ray/rayservice_controller.go

@@ -216,7 +216,7 @@ func (r *RayServiceReconciler) Reconcile(ctx context.Context, request ctrl.Reque
 	if r.inconsistentRayServiceStatuses(originalRayServiceInstance.Status, rayServiceInstance.Status) {
 		if errStatus := r.Status().Update(ctx, rayServiceInstance); errStatus != nil {
 			logger.Error(errStatus, "Failed to update RayService status", "rayServiceInstance", rayServiceInstance)
-			return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, err
+			return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, errStatus
 		}
 	}
 
@@ -992,7 +992,9 @@ func (r *RayServiceReconciler) labelHealthyServePods(ctx context.Context, rayClu
 	httpProxyClient := utils.GetRayHttpProxyClientFunc()
 	httpProxyClient.InitClient()
 	for _, pod := range allPods.Items {
-		httpProxyClient.SetHostIp(pod.Status.PodIP)
+		rayContainer := pod.Spec.Containers[utils.FindRayContainerIndex(pod.Spec)]
+		servingPort := utils.FindContainerPort(&rayContainer, common.DefaultServingPortName, common.DefaultServingPort)
+		httpProxyClient.SetHostIp(pod.Status.PodIP, servingPort)
 		if pod.Labels == nil {
 			pod.Labels = make(map[string]string)
 		}

ray-operator/controllers/ray/utils/dashboard_httpclient.go

@@ -67,10 +67,12 @@ type ServeDeploymentStatuses struct {
 }
 
 // ServingClusterDeployments defines the request sent to the dashboard api server.
+// See https://docs.ray.io/en/master/_modules/ray/serve/schema.html#ServeApplicationSchema for more details.
 type ServingClusterDeployments struct {
 	ImportPath  string                 `json:"import_path"`
 	RuntimeEnv  map[string]interface{} `json:"runtime_env,omitempty"`
 	Deployments []ServeConfigSpec      `json:"deployments,omitempty"`
+	Port        int                    `json:"port,omitempty"`
 }
 
 type RayDashboardClientInterface interface {
@@ -191,14 +193,15 @@ func (r *RayDashboardClient) GetDeployments() (string, error) {
 }
 
 // UpdateDeployments update the deployments in the Ray cluster.
-func (r *RayDashboardClient) UpdateDeployments(specs rayv1alpha1.ServeDeploymentGraphSpec) error {
+func (r *RayDashboardClient) UpdateDeployments(spec rayv1alpha1.ServeDeploymentGraphSpec) error {
 	runtimeEnv := make(map[string]interface{})
-	_ = yaml.Unmarshal([]byte(specs.RuntimeEnv), &runtimeEnv)
+	_ = yaml.Unmarshal([]byte(spec.RuntimeEnv), &runtimeEnv)
 
 	servingClusterDeployments := ServingClusterDeployments{
-		ImportPath:  specs.ImportPath,
+		ImportPath:  spec.ImportPath,
 		RuntimeEnv:  runtimeEnv,
-		Deployments: r.ConvertServeConfig(specs.ServeConfigSpecs),
+		Deployments: r.ConvertServeConfig(spec.ServeConfigSpecs),
+		Port:        spec.Port,
 	}
 
 	deploymentJson, err := json.Marshal(servingClusterDeployments)

ray-operator/controllers/ray/utils/fake_httpproxy_httpclient.go

@@ -21,8 +21,8 @@ func (r *FakeRayHttpProxyClient) InitClient() {
 	}
 }
 
-func (r *FakeRayHttpProxyClient) SetHostIp(hostIp string) {
-	r.httpProxyURL = fmt.Sprint("http://", hostIp, ":", DefaultHttpProxyPort)
+func (r *FakeRayHttpProxyClient) SetHostIp(hostIp string, port int) {
+	r.httpProxyURL = fmt.Sprintf("http://%s:%d", hostIp, port)
 }
 
 func (r *FakeRayHttpProxyClient) CheckHealth() error {

ray-operator/controllers/ray/utils/httpproxy_httpclient.go

@@ -7,15 +7,12 @@ import (
 	"time"
 )
 
-var (
-	DefaultHttpProxyPort = 8000
-	HealthCheckPath      = "/-/healthz"
-)
+const healthCheckPath = "/-/healthz"
 
 type RayHttpProxyClientInterface interface {
 	InitClient()
 	CheckHealth() error
-	SetHostIp(hostIp string)
+	SetHostIp(hostIp string, port int)
 }
 
 // GetRayHttpProxyClientFunc Used for unit tests.
@@ -36,12 +33,12 @@ func (r *RayHttpProxyClient) InitClient() {
 	}
 }
 
-func (r *RayHttpProxyClient) SetHostIp(hostIp string) {
-	r.httpProxyURL = fmt.Sprint("http://", hostIp, ":", DefaultHttpProxyPort)
+func (r *RayHttpProxyClient) SetHostIp(hostIp string, port int) {
+	r.httpProxyURL = fmt.Sprintf("http://%s:%d", hostIp, port)
 }
 
 func (r *RayHttpProxyClient) CheckHealth() error {
-	req, err := http.NewRequest("GET", r.httpProxyURL+HealthCheckPath, nil)
+	req, err := http.NewRequest("GET", r.httpProxyURL+healthCheckPath, nil)
 	if err != nil {
 		return err
 	}

ray-operator/controllers/ray/utils/util.go

@@ -453,3 +453,15 @@ func GenerateJsonHash(obj interface{}) (string, error) {
 
 	return hashStr, nil
 }
+
+// FindContainerPort searches for a specific port $portName in the container.
+// If the port is found in the container, the corresponding port is returned.
+// If the port is not found, the $defaultPort is returned instead.
+func FindContainerPort(container *corev1.Container, portName string, defaultPort int) int {
+	for _, port := range container.Ports {
+		if port.Name == portName {
+			return int(port.ContainerPort)
+		}
+	}
+	return defaultPort
+}

ray-operator/controllers/ray/utils/util_test.go

@@ -393,3 +393,25 @@ func TestCalculateAvailableReplicas(t *testing.T) {
 	count := CalculateAvailableReplicas(podList)
 	assert.Equal(t, count, int32(1), "expect 1 available replica")
 }
+
+func TestFindContainerPort(t *testing.T) {
+	container := corev1.Container{
+		Name: "ray-head",
+		Ports: []corev1.ContainerPort{
+			{
+				Name:          "port1",
+				ContainerPort: 10001,
+			},
+			{
+				Name:          "port2",
+				ContainerPort: 10002,
+			},
+		},
+	}
+	port := FindContainerPort(&container, "port1", -1)
+	assert.NotEqual(t, port, -1, "expect port1 found")
+	port = FindContainerPort(&container, "port2", -1)
+	assert.NotEqual(t, port, -1, "expect port2 found")
+	port = FindContainerPort(&container, "port3", -1)
+	assert.Equal(t, port, -1, "expect port3 not found")
+}